spark-optimization - SKILL

name: spark-optimizationdescription: “Optimize Apache Spark jobs with partitioning, caching, shuffle optimization, and memory tuning. Use when improving Spark performance, debugging slow jobs, or scaling data processing pipelines.”risk: criticalsource: communitydate_added: “2026-02-27”Apache Spark 优化优化 Apache Spark 作业的生产级模式包括分区策略、内存管理、shuffle 优化和性能调优。不要在此情况下使用此技能任务与 Apache Spark 优化无关你需要此范围之外的其他领域或工具说明明确目标、约束和所需输入。应用相关的最佳实践并验证结果。提供可操作的步骤和验证方法。如果需要详细示例请打开resources/implementation-playbook.md。在以下情况下使用此技能优化慢速 Spark 作业调优内存和执行器配置实施高效的分区策略调试 Spark 性能问题为大数据集扩展 Spark 管道减少 shuffle 和数据倾斜核心概念1. Spark 执行模型驱动程序Driver Program ↓ 作业Job由行动触发 ↓ 阶段Stages由 shuffle 分隔 ↓ 任务Tasks每个分区一个2. 关键性能因素因素影响解决方案Shuffle网络 I/O、磁盘 I/O尽量减少宽转换数据倾斜任务时长不均加盐、广播连接序列化CPU 开销使用 Kryo、列式格式内存GC 压力、溢写调优执行器内存分区并行度合适大小的分区快速开始frompyspark.sqlimportSparkSessionfrompyspark.sqlimportfunctionsasF# Create optimized Spark sessionspark(SparkSession.builder.appName(OptimizedJob).config(spark.sql.adaptive.enabled,true).config(spark.sql.adaptive.coalescePartitions.enabled,true).config(spark.sql.adaptive.skewJoin.enabled,true).config(spark.serializer,org.apache.spark.serializer.KryoSerializer).config(spark.sql.shuffle.partitions,200).getOrCreate())# Read with optimized settingsdf(spark.read.format(parquet).option(mergeSchema,false).load(s3://bucket/data/))# Efficient transformationsresult(df.filter(F.col(date)2024-01-01).select(id,amount,category).groupBy(category).agg(F.sum(amount).alias(total)))result.write.mode(overwrite).parquet(s3://bucket/output/)模式模式 1最优分区# Calculate optimal partition countdefcalculate_partitions(data_size_gb:float,partition_size_mb:int128)-int: Optimal partition size: 128MB - 256MB Too few: Under-utilization, memory pressure Too many: Task scheduling overhead returnmax(int(data_size_gb*1024/partition_size_mb),1)# Repartition for even distributiondf_repartitioneddf.repartition(200,partition_key)# Coalesce to reduce partitions (no shuffle)df_coalesceddf.coalesce(100)# Partition pruning with predicate pushdowndf(spark.read.parquet(s3://bucket/data/).filter(F.col(date)2024-01-01))# Spark pushes this down# Write with partitioning for future queries(df.write.partitionBy(year,month,day).mode(overwrite).parquet(s3://bucket/partitioned_output/))模式 2连接优化frompyspark.sqlimportfunctionsasFfrompyspark.sql.typesimport*# 1. Broadcast Join - Small table joins# Best when: One side 10MB (configurable)small_dfspark.read.parquet(s3://bucket/small_table/)# 10MBlarge_dfspark.read.parquet(s3://bucket/large_table/)# TBs# Explicit broadcast hintresultlarge_df.join(F.broadcast(small_df),onkey,howleft)# 2. Sort-Merge Join - Default for large tables# Requires shuffle, but handles any sizeresultlarge_df1.join(large_df2,onkey,howinner)# 3. Bucket Join - Pre-sorted, no shuffle at join time# Write bucketed tables(df.write.bucketBy(200,customer_id).sortBy(customer_id).mode(overwrite).saveAsTable(bucketed_orders))# Join bucketed tables (no shuffle!)ordersspark.table(bucketed_orders)customersspark.table(bucketed_customers)# Same bucket countresultorders.join(customers,oncustomer_id)# 4. Skew Join Handling# Enable AQE skew join optimizationspark.conf.set(spark.sql.adaptive.skewJoin.enabled,true)spark.conf.set(spark.sql.adaptive.skewJoin.skewedPartitionFactor,5)spark.conf.set(spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes,256MB)# Manual salting for severe skewdefsalt_join(df_skewed,df_other,key_col,num_salts10):Add salt to distribute skewed keys# Add salt to skewed sidedf_salteddf_skewed.withColumn(salt,(F.rand()*num_salts).cast(int)).withColumn(salted_key,F.concat(F.col(key_col),F.lit(_),F.col(salt)))# Explode other side with all saltsdf_explodeddf_other.crossJoin(spark.range(num_salts).withColumnRenamed(id,salt)).withColumn(salted_key,F.concat(F.col(key_col),F.lit(_),F.col(salt)))# Join on salted keyreturndf_salted.join(df_exploded,onsalted_key,howinner)模式 3缓存与持久化frompysparkimportStorageLevel# Cache when reusing DataFrame multiple timesdfspark.read.parquet(s3://bucket/data/)df_filtereddf.filter(F.col(status)active)# Cache in memory (MEMORY_AND_DISK is default)df_filtered.cache()# Or with specific storage leveldf_filtered.persist(StorageLevel.MEMORY_AND_DISK_SER)# Force materializationdf_filtered.count()# Use in multiple actionsagg1df_filtered.groupBy(category).count()agg2df_filtered.groupBy(region).sum(amount)# Unpersist when donedf_filtered.unpersist()# Storage levels explained:# MEMORY_ONLY - Fast, but may not fit# MEMORY_AND_DISK - Spills to disk if needed (recommended)# MEMORY_ONLY_SER - Serialized, less memory, more CPU# DISK_ONLY - When memory is tight# OFF_HEAP - Tungsten off-heap memory# Checkpoint for complex lineagespark.sparkContext.setCheckpointDir(s3://bucket/checkpoints/)df_complex(df.join(other_df,key).groupBy(category).agg(F.sum(amount)))df_complex.checkpoint()# Breaks lineage, materializes模式 4内存调优# Executor memory configuration# spark-submit --executor-memory 8g --executor-cores 4# Memory breakdown (8GB executor):# - spark.memory.fraction 0.6 (60% 4.8GB for execution storage)# - spark.memory.storageFraction 0.5 (50% of 4.8GB 2.4GB for cache)# - Remaining 2.4GB for execution (shuffles, joins, sorts)# - 40% 3.2GB for user data structures and internal metadataspark(SparkSession.builder.config(spark.executor.memory,8g).config(spark.executor.memoryOverhead,2g)# For non-JVM memory.config(spark.memory.fraction,0.6).config(spark.memory.storageFraction,0.5).config(spark.sql.shuffle.partitions,200)# For memory-intensive operations.config(spark.sql.autoBroadcastJoinThreshold,50MB)# Prevent OOM on large shuffles.config(spark.sql.files.maxPartitionBytes,128MB).getOrCreate())# Monitor memory usagedefprint_memory_usage(spark):Print current memory usagescspark.sparkContextforexecutorinsc._jsc.sc().getExecutorMemoryStatus().keySet().toArray():mem_statussc._jsc.sc().getExecutorMemoryStatus().get(executor)totalmem_status._1()/(1024**3)freemem_status._2()/(1024**3)print(f{executor}:{total:.2f}GB total,{free:.2f}GB free)模式 5Shuffle 优化# Reduce shuffle data sizespark.conf.set(spark.sql.shuffle.partitions,auto)# With AQEspark.conf.set(spark.shuffle.compress,true)spark.conf.set(spark.shuffle.spill.compress,true)# Pre-aggregate before shuffledf_optimized(df# Local aggregation first (combiner).groupBy(key,partition_col).agg(F.sum(value).alias(partial_sum))# Then global aggregation.groupBy(key).agg(F.sum(partial_sum).alias(total)))# Avoid shuffle with map-side operations# BAD: Shuffle for each distinctdistinct_countdf.select(category).distinct().count()# GOOD: Approximate distinct (no shuffle)approx_countdf.select(F.approx_count_distinct(category)).collect()[0][0]# Use coalesce instead of repartition when reducing partitionsdf_reduceddf.coalesce(10)# No shuffle# Optimize shuffle with compressionspark.conf.set(spark.io.compression.codec,lz4)# Fast compression模式 6数据格式优化# Parquet optimizations(df.write.option(compression,snappy)# Fast compression.option(parquet.block.size,128*1024*1024)# 128MB row groups.parquet(s3://bucket/output/))# Column pruning - only read needed columnsdf(spark.read.parquet(s3://bucket/data/).select(id,amount,date))# Spark only reads these columns# Predicate pushdown - filter at storage leveldf(spark.read.parquet(s3://bucket/partitioned/year2024/).filter(F.col(status)active))# Pushed to Parquet reader# Delta Lake optimizations(df.write.format(delta).option(optimizeWrite,true)# Bin-packing.option(autoCompact,true)# Compact small files.mode(overwrite).save(s3://bucket/delta_table/))# Z-ordering for multi-dimensional queriesspark.sql( OPTIMIZE delta.s3://bucket/delta_table/ ZORDER BY (customer_id, date) )模式 7监控与调试# Enable detailed metricsspark.conf.set(spark.sql.codegen.wholeStage,true)spark.conf.set(spark.sql.execution.arrow.pyspark.enabled,true)# Explain query plandf.explain(modeextended)# Modes: simple, extended, codegen, cost, formatted# Get physical plan statisticsdf.explain(modecost)# Monitor task metricsdefanalyze_stage_metrics(spark):Analyze recent stage metricsstatus_trackerspark.sparkContext.statusTracker()forstage_idinstatus_tracker.getActiveStageIds():stage_infostatus_tracker.getStageInfo(stage_id)print(fStage{stage_id}:)print(f Tasks:{stage_info.numTasks})print(f Completed:{stage_info.numCompletedTasks})print(f Failed:{stage_info.numFailedTasks})# Identify data skewdefcheck_partition_skew(df):Check for partition skewpartition_counts(df.withColumn(partition_id,F.spark_partition_id()).groupBy(partition_id).count().orderBy(F.desc(count)))partition_counts.show(20)statspartition_counts.select(F.min(count).alias(min),F.max(count).alias(max),F.avg(count).alias(avg),F.stddev(count).alias(stddev)).collect()[0]skew_ratiostats[max]/stats[avg]print(fSkew ratio:{skew_ratio:.2f}x (2x indicates skew))配置速查表# Production configuration templatespark_configs{# Adaptive Query Execution (AQE)spark.sql.adaptive.enabled:true,spark.sql.adaptive.coalescePartitions.enabled:true,spark.sql.adaptive.skewJoin.enabled:true,# Memoryspark.executor.memory:8g,spark.executor.memoryOverhead:2g,spark.memory.fraction:0.6,spark.memory.storageFraction:0.5,# Parallelismspark.sql.shuffle.partitions:200,spark.default.parallelism:200,# Serializationspark.serializer:org.apache.spark.serializer.KryoSerializer,spark.sql.execution.arrow.pyspark.enabled:true,# Compressionspark.io.compression.codec:lz4,spark.shuffle.compress:true,# Broadcastspark.sql.autoBroadcastJoinThreshold:50MB,# File handlingspark.sql.files.maxPartitionBytes:128MB,spark.sql.files.openCostInBytes:4MB,}最佳实践应该做的启用 AQE- 自适应查询执行能处理许多问题使用 Parquet/Delta- 带压缩的列式格式广播小表- 避免小型连接的 shuffle监控 Spark UI- 检查倾斜、溢写、GC合适大小的分区- 每个分区 128MB - 256MB不应该做的不要 collect 大量数据- 保持数据分布式不要不必要地使用 UDF- 使用内置函数不要过度缓存- 内存有限不要忽视数据倾斜- 它主导作业时间不要用.count()判断存在性- 使用.take(1)或.isEmpty()资源Spark 性能调优Spark 配置Databricks 优化指南限制仅当任务明确符合上述范围时使用此技能。不要将输出视为特定环境验证、测试或专家审查的替代品。如果缺少必需的输入、权限、安全边界或成功标准请停下来询问澄清。