<jupyter_start><jupyter_text>5일차 5교시 - Spark Bucketing> 스파크에서 버킷팅 기법이란 조인 혹은 집계 연산 시에 가장 성능을 떨어뜨리는 셔플을 피하기 위해 저장 시에 예상되는 파티션 그룹 키를 기반으로 미리 그룹핑 해서 저장해 두는 방법을 말합니다. spark.sql.sources.bucketing.enabled 를 통해 제어할 수 있고 기본 설정은 true 입니다. 결국 저장 시에 추가적인 고민과 시간이 필요하다는 의미이며, 그렇게 해서라도 충분히 리소스 사용을 상쇄할 수 있다고 판단되는 경우에만 사용하는 것을 추천 드립니다. Write Once Read Many 인 경우가 그러합니다. 목차* [1. '버킷팅'이란?](1.-'버킷팅'이란?)* [2. 언제 '버킷팅'을 적용하는가?](2.-언제-'버킷팅'을-적용하는가?)* [3. '버킷팅' 다루기](3.-'버킷팅'-다루기) 참고자료  * https://luminousmen.com/post/the-5-minute-guide-to-using-bucketing-in-pyspark<jupyter_code>from pyspark.sql import *
from pyspark.sql.functions import *
from pyspark.sql.types import *
from IPython.display import display, display_pretty, clear_output, JSON

spark = (
    SparkSession
    .builder
    .config("spark.sql.session.timeZone", "Asia/Seoul")
    .getOrCreate()
)

# 노트북에서 테이블 형태로 데이터 프레임 출력을 위한 설정을 합니다
spark.conf.set("spark.sql.repl.eagerEval.enabled", True) # display enabled
spark.conf.set("spark.sql.repl.eagerEval.truncate", 100) # display output columns size

# 공통 데이터 위치
home_jovyan = "/home/jovyan"
work_data = f"{home_jovyan}/work/data"
work_dir=!pwd
work_dir = work_dir[0]

# 로컬 환경 최적화
spark.conf.set("spark.sql.shuffle.partitions", 5) # the number of partitions to use when shuffling data for joins or aggregations.
spark.conf.set("spark.sql.streaming.forceDeleteTempCheckpointLocation", "true")
spark<jupyter_output>21/08/21 09:04:00 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).<jupyter_text>1. '버킷팅'이란?> 하둡 기반의 파일 저장구조를 활용하는 스파크의 경우 세컨더리 인덱스 지원이 어렵기 때문에 '파티셔닝'을 통해 필터하는 것에는 한계가 있기 마련입니다. 이러한 문제를 해결하기 위해 버킷팅이라는 기법을 활용할 수 있는데, 특정 키와 파티션 수를 기준으로 대상 파티션 경로 내에 또 다른 파티션 파일 블록의 저장을 통해 읽기 성능을 최적화 시킬 수 있는데, 반드시 하이브/스파크 테이블 형태로만 저장된다는 점에 유의해야만 합니다.이는   2. 언제 '버킷팅'을 적용하는가?* 디멘젼과 같은 누적 형식의 테이블의 경우 셔플이 발생하는 경우  * 조인 혹은 집계 대상이 되는 모든 데이터가 노드 간에 전달이 되기 때문에 네트워크 및 I/O 리소스를 많이 잡아 먹어 성능에 큰 영향을 미칩니다* 대용량 데이터의 셔플에 의한 익스큐터 노드의 저장데이터가 커지는 경우를 회피  * 익스큐터가 수행되는 노드의 임시 경로에 많은 데이터를 담게되므로 물리 노드에 부하를 주게되어 해당 노드에서 수행되는 다른 작업에도 영향이 있으며 클러스터 전체적인 성능 저하를 가져올 수 있습니다  3. '버킷팅' 다루기  3.1. 그대로 저장 시에는 전체 셔플이 발생합니다.![bucket1](images/bucket-1.png)<jupyter_code>s1 = spark.read.parquet("source/t1")
s2 = spark.read.parquet("source/t2")
c12 = s1.join(s2, [s1.make == s2.make]).select(s1.make, s1.model, s1.engine_size, s1.registration, s2.sale_price)
c12.printSchema()
# mode("append")
# mode("overwrite")
c1 = s1.count()
c2 = s2.count()
print(c1, c2)
c12.explain()
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1*1024*1024)
c12.where("sale_price > 2000.0").show(10)<jupyter_output>== Physical Plan ==
*(2) Project [make#1, model#2, engine_size#3, registration#0, sale_price#11]
+- *(2) BroadcastHashJoin [make#1], [make#8], Inner, BuildLeft, false
   :- BroadcastExchange HashedRelationBroadcastMode(List(input[1, string, false]),false), [id=#87]
   :  +- *(1) Filter isnotnull(make#1)
   :     +- *(1) ColumnarToRow
   :        +- FileScan parquet [registration#0,make#1,model#2,engine_size#3] Batched: true, DataFilters: [isnotnull(make#1)], Format: Parquet, Location: InMemoryFileIndex[file:/home/jovyan/work/lgde-spark-troubleshoot/source/t1], PartitionFilters: [], PushedFilters: [IsNotNull(make)], ReadSchema: struct<registration:string,make:string,model:string,engine_size:decimal(38,18)>
   +- *(2) Filter isnotnull(make#8)
      +- *(2) ColumnarToRow
         +- FileScan parquet [make#8,sale_price#11] Batched: true, DataFilters: [isnotnull(make#8)], Format: Parquet, Location: InMemoryFileIndex[file:/home/jovyan/work/lgde-spark-troubleshoot/source/t2], PartitionFilters[...]<jupyter_text>3.2. 파티션으로 저장 시에는 한쪽 파티션은 필터를 통한 잇점이 있습니다![bucket2](images/bucket-2.png)<jupyter_code>s1.write.mode("overwrite").partitionBy("make").parquet("target/troubleshoot5/model")
s2.write.mode("overwrite").partitionBy("make").parquet("target/troubleshoot5/price")
model = spark.read.parquet("target/troubleshoot5/model")
price = spark.read.parquet("target/troubleshoot5/price")
model_price = model.join(price, [model.make == price.make]).select(model.make, model.model, model.engine_size, model.registration, price.sale_price)
model_price.explain()
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1*1024*1024)
model_price.where("sale_price > 2000.0").show(10)<jupyter_output>== Physical Plan ==
*(5) Project [make#89, model#87, engine_size#88, registration#86, sale_price#96]
+- *(5) SortMergeJoin [make#89], [make#97], Inner
   :- *(2) Sort [make#89 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(make#89, 5), ENSURE_REQUIREMENTS, [id=#228]
   :     +- *(1) ColumnarToRow
   :        +- FileScan parquet [registration#86,model#87,engine_size#88,make#89] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/home/jovyan/work/lgde-spark-troubleshoot/target/troubleshoot5/model], PartitionFilters: [isnotnull(make#89)], PushedFilters: [], ReadSchema: struct<registration:string,model:string,engine_size:decimal(38,18)>
   +- *(4) Sort [make#97 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(make#97, 5), ENSURE_REQUIREMENTS, [id=#236]
         +- *(3) ColumnarToRow
            +- FileScan parquet [sale_price#96,make#97] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/home/jovyan/[...]<jupyter_text>3.3. 버킷 저장 시에는 셔플이 발생하지 않습니다.![bucket3](images/bucket-3.png)<jupyter_code>numBuckets = 10
# 이전에 삭제되지 않은 임시 경로가 있다면 삭제합니다.
!rm -rf /home/jovyan/work/spark-warehouse/model
!rm -rf /home/jovyan/work/spark-warehouse/price
s1.write.mode("overwrite").bucketBy(numBuckets, "make").sortBy("make").saveAsTable("model")
s2.write.mode("overwrite").bucketBy(numBuckets, "make").sortBy("make").saveAsTable("price")
t1 = spark.sql("select * from model")
t2 = spark.sql("select * from price")
t12 = t1.join(t2, [t1.make == t2.make]).select(t1.make, t1.model, t1.engine_size, t1.registration, t2.sale_price)
t12.explain()
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1*1024*1024)
t12.where("sale_price > 2000.0").show(10)<jupyter_output>== Physical Plan ==
*(3) Project [make#154, model#155, engine_size#156, registration#153, sale_price#164]
+- *(3) SortMergeJoin [make#154], [make#161], Inner
   :- *(1) Sort [make#154 ASC NULLS FIRST], false, 0
   :  +- *(1) Filter isnotnull(make#154)
   :     +- *(1) ColumnarToRow
   :        +- FileScan parquet default.model[registration#153,make#154,model#155,engine_size#156] Batched: true, DataFilters: [isnotnull(make#154)], Format: Parquet, Location: InMemoryFileIndex[file:/home/jovyan/work/lgde-spark-troubleshoot/spark-warehouse/model], PartitionFilters: [], PushedFilters: [IsNotNull(make)], ReadSchema: struct<registration:string,make:string,model:string,engine_size:decimal(38,18)>, SelectedBucketsCount: 10 out of 10
   +- *(2) Sort [make#161 ASC NULLS FIRST], false, 0
      +- *(2) Filter isnotnull(make#161)
         +- *(2) ColumnarToRow
            +- FileScan parquet default.price[make#161,sale_price#164] Batched: true, DataFilters: [isnotnull(make#161)], Format: Parquet, Loc[...]
{'fraction_non_alphanumeric': 0.15045947058016734, 'fraction_numerical': 0.04265532848717597, 'mean_word_length': 5.148397976391231, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 21, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '648987', 'license': 'Apache-2.0', 'n_tokens_mistral': 3628, 'n_tokens_neox': 3847, 'n_words': 629, 'path': 'day3/notebooks/lgde-spark-troubleshoot/lgde-spark-troubleshoot-5-bucket.ipynb', 'repo_name': 'psyoblade/data-engineering-advanced-training', 'size': '7291'}