동적 파티션 정리

Spark3.0 이상에는 동적 파티션 정리(DPP)가 포함되어 있습니다. 동적 파티션 정리는 데이터를 읽을 때 불필요한 파티션의 스캔을 Spark 방지하는의 최적화 기법입니다. 다음은 DPP에 대해 알아야 할 몇 가지 주요 사항입니다.

• 쿼리 필터 및 조건자에서 요청된 파티션 값을 검사하고 쿼리를 충족하는 데 필요한 파티션을 결정합니다. 불필요한 것으로 간주되는 모든 파티션은 자동으로 투명하게 정리됩니다.

• DPP는 적용 가능한 데이터가 없는 파티션을 건너뛰어 처리 시간과 리소스 사용률을 줄입니다. 이렇게 Spark 하면 관련 파티션에만 집중할 수 있습니다.

• 삽입 또는 증분 로드를 통해 추가되는 정적 파티션과 동적으로 생성된 파티션 모두에서 작동합니다.는 새 파티션을 Spark 인식하고 동적 정리를 계속 적용할 수 있습니다.

• DPP는 개발자에게 완전히 투명하거나 보이지 않습니다. DPP를 활성화하는 데 특별한 코딩이 필요하지 않습니다. 쿼리 계획 생성 중에 최적화로 백그라운드에서 자동으로 발생합니다.

다음은 DPP가 효율적으로 작동하도록 하기 위한 몇 가지 모범 사례입니다.

• Spark 데이터 프레임 작업 초기에 필터를 적용하여 조건자 푸시다운을 사용합니다. 이렇게 하면 파티션 메타데이터Spark를 사용하여 파티션을 조기에 제거할 수 있습니다.

• 를 ANALYZE TABLE 자주 실행하여 데이터에 대한 통계를 수집합니다. 이렇게 하면 무시할 수 있는 파티션Spark을 보다 정확하게 결정하는 데 도움이 되는 열 수준 통계가 줄어듭니다.

• 데이터를 과도하게 분할하지 마세요. 파티션이 너무 많으면 통계를 수집할 때 드라이버 노드에 과부하가 걸릴 수 있습니다. 각 대형 테이블에 대해 10~100개의 파티션을 목표로 합니다.

• 조인 전 데이터 프레임을 재분할합니다. 이렇게 하면 모든 데이터를 이동해야 하는 셔플 조인을 방지하고 읽는 데이터의 양을 추가로 최적화할 수 있습니다.

• 조인되는 여러 테이블에서 일관된 파티션 열 유형 및 이름 지정을 사용합니다. 이렇게 하면 조인 최적화를 위해 파티션을 Spark 더 잘 일치시킬 수 있습니다.

• 를 사용하여 쿼리를 테스트EXPLAIN하여 DPP가 적용되고 있는지 확인하고 추가 튜닝이 필요한지 확인합니다.

스타 스키마에서 테이블은 팩트 테이블과 차원 테이블의 두 가지 주요 유형으로 나뉩니다. 차원 테이블은 팩트 테이블보다 훨씬 작은 경향이 있습니다. 팩트 테이블을 차원 테이블에 조인할 때 DPP는 쿼리 계획을 최적화합니다. 차원 테이블에 적용되는 모든 필터에서 하위 쿼리를 생성합니다. 이 하위 쿼리를 브로드캐스트하고이 하위 쿼리에서 해시 테이블을 빌드합니다. 그런 다음 팩트 테이블 데이터를 읽기 전에 팩트 테이블의 스캔 단계에 해시 테이블을 적용합니다. 이렇게 하면 DPP가 더 큰 팩트 테이블에서 읽어야 하는 데이터의 양을 줄일 수 있습니다.

다음 예제 쿼리는 실행 중인 DPP를 보여줍니다. 쿼리는 국가(인도)에서 주문 수를 가져오고 팩트 테이블()과 차원 테이블(fact_orders) 간의 내부 조인을 포함합니다nation. fact_orders 테이블은 열 로 분할됩니다o_nationkey.

- "select n.n_name as country, count(1) as no_of_orders from fact_orders o join nation n on o.o_nationkey = n.n_nationkey where n.n_name = 'INDIA' group by n.n_name"  

다음은 EXPLAIN 계획에 사용되는 단계입니다.

1. 더 작은 차원 테이블(nation)을 스캔하고 열별로 필터링합니다n_name = 'INDIA'.

2. 이전 단계의 결과를 브로드캐스트합니다.

3. 첫 번째 단계의 결과를 필터링하는 하위 쿼리를 생성합니다.

4. 전체 테이블 스캔 대신 필요한 팩트 테이블 파티션만 스캔PartitionFilter하도록 로 푸시다운합니다.

다음은이 DPP 최적화 쿼리에 대한 EXPLAIN 계획입니다.

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(4) HashAggregate(keys=[], functions=[count(1)], output=[count#208L])
   +- ShuffleQueryStage 3
      +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#353]
         +- *(3) HashAggregate(keys=[], functions=[partial_count(1)], output=[count#212L])
            +- *(3) HashAggregate(keys=[n_name#31], functions=[], output=[])
               +- ShuffleQueryStage 1
                  +- Exchange hashpartitioning(n_name#31, 36), ENSURE_REQUIREMENTS, [id=#315]
                     +- *(2) HashAggregate(keys=[n_name#31], functions=[], output=[n_name#31])
                        +- *(2) Project [n_name#31]
                           +- *(2) BroadcastHashJoin [cast(o_nationkey#145 as bigint)], [n_nationkey#32L], Inner, BuildRight, false
                              :- *(2) ColumnarToRow
                              :  +- FileScan parquet [o_nationkey#145] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[s3://aws-spark-tuning/fact_orders], PartitionFilters: [isnotnull(o_nationkey#145), dynamicpruningexpression(cast(o_nationkey#145 as bigint) IN dynamicp..., PushedFilters: [], ReadSchema: struct<>
                              :        +- SubqueryBroadcast dynamicpruning#210, 0, [n_nationkey#32L], [id=#200]
                              :           +- OutputAdapter [n_name#31, n_nationkey#32L]
                              :              +- AdaptiveSparkPlan isFinalPlan=true
                              :                 +- BroadcastQueryStage 2
                              :                    +- ReusedExchange [n_name#31, n_nationkey#32L], BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#233]
                              +- BroadcastQueryStage 0
                                 +- BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#233]
                                    +- *(1) Filter ((isnotnull(n_name#31) AND (n_name#31 = INDIA)) AND isnotnull(n_nationkey#32L))
                                       +- FileScan json [n_name#31,n_nationkey#32L] Batched: false, DataFilters: [isnotnull(n_name#31), (n_name#31 = INDIA), isnotnull(n_nationkey#32L)], Format: JSON, Location: InMemoryFileIndex[s3://aws-spark-tuning/input/demo/json/nation], PartitionFilters: [], PushedFilters: [IsNotNull(n_name), EqualTo(n_name,INDIA), IsNotNull(n_nationkey)], ReadSchema: struct<n_name:string,n_nationkey:bigint>

o_nationkey 열에 직접 필터가 추가되지 않았더라도 DPP 기능으로 인해는 전체 테이블 대신 필요한 파티션만 Spark 자동으로 스캔합니다.