spark-scheduling

Spark-Scheduling?

개인적으로 Spark를 사용하는 이유는 cluster-mode를 사용하기 위해라고 생각한다
Yarn등을 이용하여 각 Driver,Excutor에게 core,memory등을 할당 하는데, 많이 할당한다고 항상 좋은 결과를 보는 것은 아니다
어디서 본지 기억이 나지 않지만, core는 1~5개, memory는 4gb이상을 사용하는 것이 좋다고 봤었다.

Cluster Scheduling

Static resource allocation

정적(고정) 자원 할당

• 명칭 그대로 최초 할당한 그대로 어플리케이션이 종료될 때 까지 자원을 점유하는 방식
• Pipe line에 따라 데이터를 처리하여 필요한 자원 사전 계산 후 할당하면 좋은 방식이다.

Dynamic resource allocation

동적 자원 할당

• 자원을 필요에 따라 동적으로 할당하는 방식

  • Spark-Shell , 혹은 web기반으로 동작하면 항상 많은 자원을 할당받아 자원 낭비가 발생할 수 있는 상황에 합리적인 선택

• 공통

  • spark.dynamicAllocation.enabled true 설정을 해야한다

• Standalone

  • spark.shuffle.service.enabled true

• Yarn

  • 모든 node manager에 spark-{version}-yarn-shuffle.jar를 class path등록해야 한다
  • yarn-site.xml에 아래와 같은 설정이 있어야 한다.(안 되어 있으면 설정 후 재시작)

  	<property>
          <name>yarn.nodemanager.aux-services</name>
          <value>mapreduce_shuffle,spark_shuffle</value>
      </property>
      <property>
          <name>yarn.nodemanager.aux-services.spark_shuffle.class</name>
          <value>org.apache.spark.network.yarn.YarnShuffleService</value>
      </property>
      <property>
          <name>spark.shuffle.service.enabled</name>
          <value>true</value>
      </property>
  

• yarn에서 별도 Shuffle Service를 실행시킨 이유는 동작 도중 Executor가 제거될 수 있기 때문

  • Mapper에 저장된 데이터가 Reducer단계에 Network를 통해 데이터를 읽어 가는데 이때 Executor가 삭제되면 해당 데이터는 삭제됨
  • 그러한 문제가 발생하지 않기 위해 Shuffle data를 따로 관리 할 수 있는 Shuffle Process를 설정하는 것

Application Scheduling

Spark Context는 기본적으로 Multi-Thread 방식으로 실행된다.
그로인하여 N+1개의 Action Function을 동시 실행해도 문제가 발생하지 않는다.

• Scheduling으로 FIFO방식으로 구동 된다
  • 후속 작업은 이전 작업이 끝날 때 까지 대기해야한다
  • Fair scheduler 방식도 가능하다
    • Round robin
  • 설정
    • Code 단 : conf.set("spark.scheduler.mode", "Fair")
    • spark-default.conf에서 설정해도 된다.

fairscheduler.xml에 poll 설정을 하여 원하는 Pool이용하는 코드를 작성할 수 있다

<allocations>
  <pool name="production">
    <schedulingMode>FAIR</schedulingMode>
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
  <pool name="test">
    <schedulingMode>FIFO</schedulingMode>
    <weight>2</weight>
    <minShare>3</minShare>
  </pool>
</allocations>

• fairscheduler.xml에 기본적으로 작성되어있는 것을 가져왔다.
  • weight : 우선순위
  • minShare : Pool의 최초 CPU Core

val conf = new SparkConf()  
  .setAppName("application_name")  
  .setMaster("local[*]")  
  .set("spark.local.ip", "localhost")  
  .set("spark.driver.host", "localhost")  
  // .set("spark.scheduler.mode", "Fair")  //이와 같이 특정 scheduler 사용가능
  .set("spark.scheduler.allocation.file" ,"file path") // 
val sc = new SparkContext(conf)
sc.setLocalProperty("spark.scheduler.pool", "production") // pool설정  
sc.setLocalProperty("spark.scheduler.pool", null) // 기본 pool설정

• 위와 같이 사용하면 scheduler mode : FAIR , 최소 CPU core : 2
• 이와 같이 N+1명의 사용자가 접근하면 사용자에 따른 자원 할당을 할 수 있다

[ Reference ]

• 빅데이터 분석을 위한 스파크2 프로그래밍
• Spark 공식홈페이지

---

개인적으로 공부한 내용 포스팅 중
잘못된 정보는 지적해주시면 좋겠습니다!