📒 Spark(19)

📌 Repartition and Coalesce

Repartition을 하는 이유

• 전체적으로 Partition의 수를 늘려 병렬성을 증가시키기 위해서이다.
• 굉장히 큰 파티션이나 Skew 파티션의 크기를 조절하기 위해서이다.
• 파티션을 분석 패턴에 맞게 재분배하기 위해서이다. -> Write once, Read many
  • 어떤 DataFrame을 특정 컬럼 기준으로 그룹핑을 하거나 필터링을 자주 하는 경우
    -> 미리 그 컬럼 기준으로 저장해 두었다면 그게 Bucketing인 것이다.

Repartition을 방식

• 2가지 방식이 존재한다.

  • repartition
  • repartitionByRange

• Shffling이 발생한다. -> 때문에 분명한 이유를 가지고 Repartition을 사용해야한다.

  • 많은 경우 repartition이 별 이유없이 사용되어 오히려 시간과 비용이 증가한다.
    -> 비슷하게 불필요한 Counting과 Distinct Counting과 Duplicate제거 비용이 발생한다.

• Column이 사용되면 균등한 Partition 크기를 보장할 수없다.

• Partition수를 줄이는 용도로는 사용불가하다.
  -> 줄이는 경우에는 Coalesce를 사용해야한다.

• repartition(numPartitions, *cols)

  • Hash 기반 Partitioning
    • repartition(5)
    • repartition(5, "city")
    • repartition(5, "city", "zipcode")
    • repartition("city")
    • repartition("city", "zipcode")

• repartitionByRange(numPartitions, *cols)

  • 지정된 컬럼 값의 범위를 기준으로 파티션을 나누는 방식이다.
  • 데이터 샘플링 기반으로 파티션을 나누기에 결과가 매번 다를 수 있다.
    • Nondeterministic
  • 사용법 자체는 앞서 repartition과 동일하다.

Coalesce 가 필요한 경우

• 파티션의 수를 줄이는 용도(늘리지 않는다.)
• Shiffling이 발생시키지 않고 로컬 파티션들을 Merge한다.
  -> 따라서 Skew 파티션을 만들어낼 수 있다.
• Column이 사용되며 균등한 파티션 크기를 보장할 수 없다.

DataFrame관련 Hint

• Spark SQL Optimizer에게 Execution Plan을 만듬에 있어서 특정한 방식을 사용하도록 제안
  -> 최적화된 방식을 변경하기 위해 사용한다.

• 두 종류의 힌트들이 존재한다.

  • Partition 관련 Hint
  • Join 관련 Hint

참고

Partition 관련 Hint

• COALESCE
• REPARTITION
• REPARTITION_BY_RANGE
• REBALANCE
  • DataFrame을 테이블로 저장할 때 아주 유용하다.
  • 파일의 크기를 최대한 비슷하게 만들어서 저장한다.(AQE가 필요함)

df1.join(df2, "id", "inner").hint("COALESCE", 3)

Join 관련 Hint

• BROADCAST, BROADCASTJOIN, MAPJOIN

  • Broadcast Join 사용 제안

• MERGE, SHUFFLE_MERGE, MERGEJOIN

  • Shuffle Merge Join 사용 제안
  • Spark의 기본 조인 전략

• SHUFFLE_HASH

  • Shuffle Hash Join 사용 제안
  • Full Outer Join에는 사용 불가

• SHUFFLE_REPLICATE_NL

  • Shuffle-and-replicate (Cross Join) Join 사용 제안

• 여러 개가 동시에 사용될 경우 BROADCAST -> SHFFLE_REPLICATE_NL 순으로 우선순위가 낮아진다.

SELECT /*+ MERGE(df2) */ *
FROM df1 JOIN df2 ON df1.order_month = df2.year_month

참고

DataFrame Hint 사용법

• Spark SQL
  -> /* + hint[,...] */

SELECT /*+ REPARTITION(3) */ * FROM TABLE
SELECT /*+ BROADCAST(table1) */ * FROM table1
JOIN table2 ON table1.key = table2.key

• DataFrame API
  -> .hint 메소드 사용

join_df = df1.join(df2, "id", "inner").hint("COALESCE", 3)
join_df = df1.join(df2.hint("broadcast"), "id", "inner").hint("COALESCE", 3)

---

📌 AQE (Adaptive Query Execution)

Spark Optimization의 역사

• Spark 1.x : Catalyst Optimizer & Tungsten Project

  • 전자는 규칙기반 최적화를 수행(Predicate pushdown, projection pushdown)
  • 후자는 기본적으로 JVM 문제없이 코드 최적화를 하려는 것(GC를 피하기 위해 직접 Off Heap 메모리 관리를 수행한다.)

• Spark 2.x : CBO(Cost-Based Opimizer)

  • 데이터 프레임 통계정보를 이용하여 효율적인 Execution plan을 생성한다.
    -> 전체 크기, 레코드 수, 컬럼별 특성 (최소/최대/히스토그램 등등)

spark.sql.shuffle.partitions

SELECT sku, SUM(price) sales
FROM order
GROUP BY sku;

• 이 쿼리에 GROUP BY문은 2개의 stage를 만들어낸다.

• spark.sql.shuffle.partitions값에 의해 Shuffling 후 Partition 수를 결정한다.
  

• 이 변수 하나로 다양한 상황의 Shuffling을 해결하기는 쉽지 않다.
  ->MapReduce 세상에서 mapreduce.job.reduces와 동일하다.

• 적은 수의 Partition은 병렬성을 낮추고 OOM과 disk spill의 가능성을 높인다.

• 많은 수의 Partition은 task scheduler와 task 생성과 관련된 오버헤드가 생기며 너무 흔한 네트워크 I/O 요청으로 병목현상을 초래한다.

이슈 해결

• 대용량 DB 분야에서는 이 문제는 잘 연구된 문제이다.
• Intel Big Data팀이 프로토타입을 개발한 후 DataBricks와 협업을 하였다.(Spark 3.0)
• 기본적인 아이디어는 parsing time 최적화와 runtime 최적화의 병행이다.
  • Parsing time 정보로 선택된 physical plan과 코드 최적화만으로는 불충분하다.
  • 특히 UDF가 많이 사용되는 경우 이 문제는 더욱 심각해진다.

AQE - Adaptive Query Execution

• 런타임 통계를 기반으로 쿼리 실행 중에 발생하는 동적 쿼리 최적화이다.

• AQE는 정확한 런타임 통계를 바탕으로 모든 최적화 결정을 내린다.

• 그렇다면 언제 이런 실행시간 통계 정보를 뽑고 최적화 방식에 변경을 줄 수 있는 최적의 시점일까?

  • 정답은 Stage이다.
  • Query -> Job -> Stage -> Task

Stage가 가장 좋은 최적화 방식 변경

• Shuffling/Broadcasting이 Job을 Stage들로 나눈다.
• 또한 이 때 중간 결과들이 materialize된다.
  -> 따라서 가장 좋은 시점이며 또한 Partition의 수와 크기 정보들도 알 수 있다.
  

AQE 이후 세상

• 앞서 Group By 쿼리 2번째 Stage 시작부에 AQEShufflingRead가 사용된다.
  

AQE가 필요한 경우들

• Dynamically coalescing (Post) shuffle partitions(Spark 3)
• Dynamically switching join strategies (Spark 3.2)
• Dynamically optimizing skew joins(Spark 3)

---

📌 Dynamically coalescing (Post) shuffle partitions

AQE의 동작 방식

• Stage Dag를 순차적으로 실행
• 매번 새로운 최적화 기회가 있는지 조사
  -> 필요하다면 다시 실행하거나 쿼리 플렌을 변경한다.

Dynamically coalescing shuffle partitions 동작 방식

• AQE의 해법은 다음과 같다.

• 내부적으로 많은 수의 파티션을 고의로 생성한다.
  -> spark.sql.adaptive.coalescePartitions.initialPartitionNum(200)

• 매 Stage가 종료될 때 필요하다면 자동으로 Coalesce를 수행한다.
  -> spark.sql.adaptive.coalescePartitions.enabled

• 설정에 따라 파티션의 크기는 최소 크기 혹은 목표 크기를 맞추려 동작한다.

  • spark.sql.adaptive.advisoryPartitionSizeInBytes
  • spark.sql.adaptive.coalescePartitions.minPartitionSize
  • 무엇을 선택할지는 spark.sql.adaptive.coalescePartitions.parallelismFirst에 의해서 결정된다.

Dynamically coalescing shuffle partitions란?

• 왜 필요한가?
  • 적당한 Partition의 크기와 수는 성능에 지대한 영향을 끼친다.
  • 너무 많은 수의 작은 Partition
    • 스케쥴러 오버헤드
    • 테스트 준비 오버헤드
    • 비효율적인 I/O (파일 시스템/네트워크)
  • 적은 수의 큰 Partition
    • GC악몽 -> OOM = Out Of Memory
    • Disk Spill
  • spark.sql.shuffle.partitions라는 하나의 변수로는 불충분하다.

Dynamically coalescing shuffle partitions 관련 변수들(3.3.1)

환경 변수 이름	기본 값	설명
spark.sql.adaptive.coalescePartitions.enabled	True	spark.sql.adaptive.enabled도 true인 경우 셔플 후 파티션 수를 동적으로 줄이며 파티션의 크기는 아래 변수 (advisoryPartitionSizeInBytes)로 맞추려 시도한다.
spark.sql.adaptive.advisoryPartitionSizeInBytes	64MB	셔플링 후 파티션 수를 줄일 떄 목표로 하는 파티션의 크기
spark.sql.adaptive.coalescePartitions.parallelismFirst	True	이 값이 True이면 병렬성 보장을 위해 위의 목표 크기가 무시되고 아래 minPartitionSize만 보장된다.
spark.sql.adaptive.coalescePartitions.initialPartitionNum	None	Coalespcing 전의 파티션 수, 없으면 spark.sql.shuffle.partitions로 설정한다.

Dynamically coalescing shuffle partitions

• Coalescing이 없는 Shuffle
  

• AQE Coalescing이 적용된 경우
  

AQE 데모 환경

• 2개의 테이블을 만들어서 3가지의 AQE 기능을 테스트하는데 사용한다.

• Table_1 (items) : 일종의 dimension 테이블

  • 30,000,000개의 레코드가 존재.
  • id와 price 필드로 구성

• Table_2(sales) : 일종의 Fact 테이블

  • 1,000,000,000개의 레코드가 존재.
  • item_id와 quantity와 date 필드로 구성

• 위 두 개의 테이블을 Spark SQL을 사용하여 생성한다.

---

📌 Dynamically switching join strategies

Dynamically switching join strategies란

• 왜 필요한가

  • Static Query Plan이 여러 이유로 BHJ(Broadcast Hash Join)기회를 놓친 경우
    • 조인대상 DataFrame들에 대한 통계 정보 부족(필터링 등등)
    • UDF가 사용된 경우

• AQE의 해법

  • Runtime 통계정보를 바탕으로 조인 전략을 변경한다.
    -> 이는 Stage들이 끝나고 조인되기 전에 다시 쿼리 플래닝을 수행한다.
  • 아래 두 가지 옵션이 존재한다.
    • Broadcast Join (추천되며 우선순위를 갖는다.)
    • Shuffle Hash Join

Dynamically switching join strategies 동작방식

1. Leaf stage 실행
2. 결과를 보고 다시 query planning(최적화)
   -> B가 필터링 후에 Broadcastr 필터링 밑으로 내려가는 경우 B로 Broadcast조인을 수행한다.
3. Static query planning보다는 성능이 떨어진다.(하지만 여전히 도움이 된다.)

spark.sql.adaptive.autoBroadcastJoinThreshold
spark.autoBroadcastJoinThreshold

Dynamically switching join 관련 환경변수(Spark 3.3.1)

환경 변수 이름	기본 값	설명
spark.sql.join.preferSortMergeJoin	True	데이터프레임 조인시 Sort Merge Join을 기본으로 사용할지 여부, 항상 Sort Merge Join을 쓴다는 의미는 아니다.
spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold	0	Hash Join시 파티션 별로 해시맵 생성에 사용가능한 최대 크기 지정. 이 값이 spark.sql.adaptive.advisorPartitionSizeInBytes보다 크고 모든 파티션 크기가 이 값보다 작다면 Hash Join을 선택하여 조인을 진행한다. spark.sql.join.preferSortMergeJoin의 값은 무시된다.
spark.sql.adaptive.autoBroadcastJoinThreshold	없음	브로드캐스트 가능한 데이터프레임의 최대 크기. 이 값을 -1로 설정하면 브로드캐스트는 사용되지 않는다. 기본값은 spark.sql.autoBroadcastJoinThreshold와 동일하며 AQE가 활성화된 경우에만 사용된다.