[Spark] 스파크 스케쥴링

Application 스케쥴링

관련하여 https://community.cloudera.com/t5/Community-Articles/Dynamic-Allocation-in-Apache-Spark/ta-p/368095  이쪽에 정리가 잘 되어 있다.

SRA(Static Resource Allocation)

spark 어플리케이션은 실행되기 전 어플리케이션에서 사용할 리소스를 미리 예약해둔다.

이 리소스의 양은 고정되어서 런타임중에는 변경이 불가능하다. 만약 할당된 것보다 더 많은 리소스가 필요하다면 실행시간이 길어지거나 실패할 수 있다.

리소스가 고정되어 있기 때문에 만약 할당된 리소스가 어플리케이션에서 사용되지 않는다면 비효율적인 리소스 활용으로 이어지게 된다.

또한 런타임중에 리소스를 조정하는것이 불가능하기 때문에 중간에 리소스 투입이 더 필요하다면 out of memory 에러가 발생할 수 있다.

http://www.riveriq.com/blogs/2018/08/dynamic-allocation-in-spark

 

DRA(Dynamic Resource Allocation)

dynamic resource allocation을 활성화시키면 런타임중에 필요에 따라 리소스가 어플리케이션에 할당된다. 이를 통해 리소스 활용도가 향상되고 리소스 활용도가 낮거나 과도하게 활용하는 것을 방지할 수 있다. dynamic resouce allocation에서의 리소스 공유 단위는 Executor이며, 아래에 나오는 모든 "리소스"단어의 단위는 Executor이다.

http://www.riveriq.com/blogs/2018/08/dynamic-allocation-in-spark

동적할당을 함으로써 생기는 장점은 아래와 같다.

1. 리소스를 효율적으로 사용할 수 있고 클러스터의 전반적인 효율성을 향상시킬 수 있다.
2. 워크로드에 따라 할당된 리소스를 확장하거나 축소할 수 있다.
3. 리소스를 효율적으로 할당하기 때문에 어플리케이션 전체 비용을 줄일 수 있다.
4. 한 클러스터에서 여러 어플리케이션이 실행될 때 공정하게 리소스를 할당받을 수 있다.

반대로 단점은 아래와 같다.

1. 스파크에서 워크로드를 지속적으로 모니터링하고 리소스 할당을 조정해야하기 때문에 추가적인 오버헤드가 발생하고, 이부분은 어플리케이션 성능에 영향을 미칠 수 있다.
2. 리소스 조정에 따른 레이턴시가 있을 수 있다.
3. 관련 속성값 구성이 추가로 필요해 어플리케이션 관리 및 배포가 더 복잡해질 수 있다.
4. 할당된 리소스가 자주 변경되면 사용할 executor에 대해서 예측불가능해진다.
5. 매 리소스 요청 및 해제 시마다 node manager와 통신해야 하기 때문에 네트워크 비용을 증가시킬 수 있다.
6. 할당된 리소스가 부족하다면 셔플 프로세스를 처리하는 yarn shuffle service가 오버로드 될 수 있고, 이는 전체 클러스터 성능 저하를 야기할 수 있다. 동적으로 할당되는 리소스 모니터링을 잘 해야한다.

관련 구성

Property Name	Default Value	Description
spark.shuffle.service.enabled	false	Enables the external shuffle service.
spark.dynamicAllocation.enabled	false	Set this to true to enable dynamic allocation.
spark.dynamicAllocation.minExecutors	0	어플리케이션에 할당해야 하는 최소 executor 수
spark.dynamicAllocation.initialExecutors	spark.dynamicAllocation.minExecutors	동적 할당이 활성화된 경우 실행할 initial executor 수 `--num-executors` (or `spark.executor.instances`) 가 설정되어 있고, 이 값보다 큰 경우에 사용됨
spark.dynamicAllocation.maxExecutors	infinity	어플리케이션에 할당해야 하는 최대 executor 수

* 참고: yarn shuffle service

https://mallikarjuna_g.gitbooks.io/spark/content/yarn/spark-yarn-YarnShuffleService.html

dynamic resource allocation을 사용하기 위해 활성화해야하는 external shuffle service. 스파크 executor가 셔플 파일을 가져오는데 사용되며, executor 하나가 fail되더라도 shuffled file은 손실되지 않는다.

https://medium.com/@rachit1arora/apache-spark-shuffle-service-there-are-more-than-one-options-c1a8e098230e

 

Job 스케쥴링

FIFO 스케쥴링

한 job의 모든 task가 종료되면 다음 job을 실행하는 First in first out 스케쥴링이다.

디폴트로 FIFO 방식으로 리소스를 할당하며, 첫 번째 정의된 작업이 사용가능한 모든 리소스에 대해 우선순위를 갖게 된다.

첫 번째 job에서 모든 리소스가 필요하지 않다면 다음 작업에서도 해당 리소스를 사용할 수 있다.

https://www.waitingforcode.com/apache-spark/fair-jobs-scheduling-apache-spark/read#FAIR_scheduling_Apache_Spark

그러나 문제될 수 있는 상황은 첫 번쨰 Job에서 실행해야 하는 task 가 많을 때 다음 Job의 task 수가 아무리 작아도 다음 job은 첫 번쨰 job이 완료되기를 기다려야 한다는 지점이다.

이러한 문제를 해결하기 위해 FAIR 스케쥴링을 사용할 수 있다.

 

FAIR 스케쥴링

Fair 스케쥴링은 라운드 로빈 방식으로 작동한다.

아래 이미지에서 볼 수 있다시피 job2는 job1이 종료될 때까지 기다릴 필요 없이 가능한 한 빨리 시작될 수 있다.

하나의 application 내에서 여러 작업의 실행시간을 최적화할 수 있는 모드 중 하나이다. FIFO 와 달리 작업 간 리소스를 공유하기 떄문에 긴 시간 실행되는 하나의 job때문에 리소스가 lock되는 상황일 피할 수 있다.

 

사용방법

`spark.scheduler.mode`를 FAIR로 설정해주면 된다.

from pyspark.sql SparkSession
from pyspark import SparkConf

conf = SparkConf()
conf.set("spark.scheduler.mode", "FAIR")
spark = SparkSession.builder.appName(...)\
		...
        .config(conf=conf)\
        ...

FAIR 모드 실행 모습

scheduler pool

https://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application

Job을 pool로 일종의 그룹화를 하여 관리하고 각 pool에 가중치 등 서로 다른 옵션 설정을 지원한다.

스케쥴링 모드를 FAIR로 설정 시 기본적으로 각 pool은 클러스터에서 동일한 FAIR 모드로 실행되지만, 각 pool 내에서의 작업은 FIFO로 수행된다.

 

 

Pool 관련 속성

`schedulingMode`

pool 대기열 내의 작업의 스케쥴링 모드이다. FIFO or FAIR

`weight`

다른 pool을 기준으로 클러스터의 pool 공유를 제어한다. 디폴트 값은 1로 모든 pool의 가중치는 1이다.

예를 들어 특정 pool에 가중치를 2로 설정하면 다른 pool들보다 2배 더 많은 리소스를 얻을 수 있다.

1000 과 같은 아주 큰 수로 설정해 pool간의 우선순위를 구현할 수도 있다. ( 가중치 1000 로 설정하면 즉 해당 pool에 있는 작업이 활성화될 때마다 항상 먼저 작업을 시작하게 된다는 의미이다.)

`minShare`

전체 가중치와 별도로 각 pool의 최소 리소스(CPU 코어) 를 설정한다. 스케쥴러는 가중치에 따라 리소스를 재분배하기 전에 항상 모든 활성 pool의 minShare를 충족해야 한다.

따라서 minShare 값은 pool 이 항상 특정 수의 리소스(CPU 코어)를 빠르게 확보할 수 있도록 보장할 수 있는 값이어야 한다. 디폴트 값은 0이다.

 

Pool 사용 방법

먼저 XML 파일을 작성해 각 Pool 을 정의해준다.

<?xml version="1.0"?>
<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>

그리고 spark config 내 `spark.scheduler.allocation.file`에 해당 파일 위치를 작성해준다. 아니면 conf 디렉토리의 fairscheduler.xml 파일에 기재해주어도 된다.

# scheduler file at local
conf.set("spark.scheduler.allocation.file", "file:///path/to/file")
# scheduler file at hdfs
conf.set("spark.scheduler.allocation.file", "hdfs:///path/to/file")

pool을 정의할 때에는 아래와 같이 어떤 pool인지 설정해주면 된다. 참고로 아래와 같이 설정하면 "test"라는 이름의 pool 이 있는지 먼저 spark.scheduler.allocation.file을 확인하고, fairscheduler.xml 파일을 확인하고, 일치하는 풀이 없다면 default pool로 이동한다.

# Assuming sc is your SparkContext variable
sc.setLocalProperty("spark.scheduler.pool", "test")

연결된 pool을 지우기 위해서는 아래와 같이 null로 설정해주면 된다. 그럼 해당 쓰레드 내에서 제출된 작업들은 디폴트 pool로 들어간다.

# Assuming sc is your SparkContext variable
sc.setLocalProperty("spark.scheduler.pool", None)

 

 

참고

https://youtu.be/BLT6eHcT-e8?si=dgDttOPjW7I40mH-

https://www.waitingforcode.com/apache-spark/fair-jobs-scheduling-apache-spark/read#FAIR_scheduling_Apache_Spark

https://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application