From 070d764c5a8d970893bf951d3825ce6c52ec2589 Mon Sep 17 00:00:00 2001 From: wehub-skill-sync Date: Mon, 13 Jul 2026 21:36:28 +0800 Subject: [PATCH] chore: import zh skill spark-optimization --- README.wehub.md | 9 ++ SKILL.md | 95 +++++++++++++ references/details.md | 321 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 425 insertions(+) create mode 100644 README.wehub.md create mode 100644 SKILL.md create mode 100644 references/details.md diff --git a/README.wehub.md b/README.wehub.md new file mode 100644 index 0000000..00928c9 --- /dev/null +++ b/README.wehub.md @@ -0,0 +1,9 @@ +# WeHub 来源说明 + +- Skill 名称：`spark-optimization` +- 中文类目：Apache Spark 作业性能调优 +- 上游仓库：`wshobson__agents` +- 上游路径：`plugins/data-engineering/skills/spark-optimization/SKILL.md` +- 上游链接：https://github.com/wshobson/agents/blob/HEAD/plugins/data-engineering/skills/spark-optimization/SKILL.md +- 本仓库为 WeHub 中文 Skill 汉化包，基于 skill 市场筛选 Top200 清单整理 +- 原作者、版权和许可证信息以上游仓库为准 diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..b47c6b4 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,95 @@ +--- +name: spark-optimization +description: 优化 Apache Spark 作业，涵盖分区、缓存、shuffle 优化与内存调优。适用于提升 Spark 性能、调试慢作业或扩缩数据处理管道时使用。 +--- + +# Apache Spark 优化 + +优化 Apache Spark 作业的生产级模式，包括分区策略、内存管理、shuffle 优化与性能调优。 + +## 何时使用此技能 + +- 优化运行缓慢的 Spark 作业 +- 调优内存与执行器配置 +- 实现高效的分区策略 +- 调试 Spark 性能问题 +- 为大数据集扩缩 Spark 管道 +- 减少 shuffle 与数据倾斜 + +## 核心概念 + +### 1. Spark 执行模型 + +``` +Driver 程序 + ↓ +作业（由 action 触发） + ↓ +阶段（以 shuffle 分隔） + ↓ +任务（每个分区一个） +``` + +### 2. 关键性能因素 + +| 因素 | 影响 | 解决方案 | +| ------------------- | ----------------------- | ------------------------------ | +| **Shuffle** | 网络 I/O、磁盘 I/O | 减少宽依赖转换 | +| **数据倾斜** | 任务执行时间不均 | 加盐、广播连接 | +| **序列化** | CPU 开销 | 使用 Kryo、列式格式 | +| **内存** | GC 压力、溢出到磁盘 | 调优执行器内存 | +| **分区数** | 并行度 | 设置合适的分区大小 | + +## 快速开始 + +```python +from pyspark.sql import SparkSession +from pyspark.sql import functions as F + +# 创建优化后的 Spark 会话 +spark = (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()) + +# 使用优化设置读取数据 +df = (spark.read + .format("parquet") + .option("mergeSchema", "false") + .load("s3://bucket/data/")) + +# 高效转换 +result = (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/") +``` + +## 详细模式与示例 + +详细模式文档位于 `references/details.md`。当上述导航层级不够用时，请阅读该文件。 + +## 最佳实践 + +### 应该做的 + +- **启用 AQE** — 自适应查询执行能处理许多问题 +- **使用 Parquet/Delta** — 列式格式并启用压缩 +- **广播小表** — 对小表连接避免 shuffle +- **监控 Spark UI** — 检查数据倾斜、溢出、GC 情况 +- **设置合适的分区大小** — 每个分区 128MB — 256MB + +### 不应做的 + +- **不要收集大数据** — 保持数据分布式 +- **不要不必要地使用 UDF** — 优先使用内置函数 +- **不要过度缓存** — 内存是有限的 +- **不要忽视数据倾斜** — 它会主导作业运行时间 +- **不要用 `.count()` 判断是否存在** — 应使用 `.take(1)` 或 `.isEmpty()` diff --git a/references/details.md b/references/details.md new file mode 100644 index 0000000..4df773a --- /dev/null +++ b/references/details.md @@ -0,0 +1,321 @@ +# spark-optimization —— 详细模式与实操示例 + +## Patterns（模式） + +### Pattern 1: Optimal Partitioning（最优分区） + +```python +# Calculate optimal partition count +def calculate_partitions(data_size_gb: float, partition_size_mb: int = 128) -> int: + """ + Optimal partition size: 128MB - 256MB + Too few: Under-utilization, memory pressure + Too many: Task scheduling overhead + """ + return max(int(data_size_gb * 1024 / partition_size_mb), 1) + +# Repartition for even distribution +df_repartitioned = df.repartition(200, "partition_key") + +# Coalesce to reduce partitions (no shuffle) +df_coalesced = df.coalesce(100) + +# Partition pruning with predicate pushdown +df = (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/")) +``` + +### Pattern 2: Join Optimization（连接优化） + +```python +from pyspark.sql import functions as F +from pyspark.sql.types import * + +# 1. Broadcast Join - Small table joins +# Best when: One side < 10MB (configurable) +small_df = spark.read.parquet("s3://bucket/small_table/") # < 10MB +large_df = spark.read.parquet("s3://bucket/large_table/") # TBs + +# Explicit broadcast hint +result = large_df.join( + F.broadcast(small_df), + on="key", + how="left" +) + +# 2. Sort-Merge Join - Default for large tables +# Requires shuffle, but handles any size +result = large_df1.join(large_df2, on="key", how="inner") + +# 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!) +orders = spark.table("bucketed_orders") +customers = spark.table("bucketed_customers") # Same bucket count +result = orders.join(customers, on="customer_id") + +# 4. Skew Join Handling +# Enable AQE skew join optimization +spark.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 skew +def salt_join(df_skewed, df_other, key_col, num_salts=10): + """Add salt to distribute skewed keys""" + # Add salt to skewed side + df_salted = df_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 salts + df_exploded = df_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 key + return df_salted.join(df_exploded, on="salted_key", how="inner") +``` + +### Pattern 3: Caching and Persistence（缓存与持久化） + +```python +from pyspark import StorageLevel + +# Cache when reusing DataFrame multiple times +df = spark.read.parquet("s3://bucket/data/") +df_filtered = df.filter(F.col("status") == "active") + +# Cache in memory (MEMORY_AND_DISK is default) +df_filtered.cache() + +# Or with specific storage level +df_filtered.persist(StorageLevel.MEMORY_AND_DISK_SER) + +# Force materialization +df_filtered.count() + +# Use in multiple actions +agg1 = df_filtered.groupBy("category").count() +agg2 = df_filtered.groupBy("region").sum("amount") + +# Unpersist when done +df_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 lineage +spark.sparkContext.setCheckpointDir("s3://bucket/checkpoints/") +df_complex = (df + .join(other_df, "key") + .groupBy("category") + .agg(F.sum("amount"))) +df_complex.checkpoint() # Breaks lineage, materializes +``` + +### Pattern 4: Memory Tuning（内存调优） + +```python +# 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 metadata + +spark = (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 usage +def print_memory_usage(spark): + """Print current memory usage""" + sc = spark.sparkContext + for executor in sc._jsc.sc().getExecutorMemoryStatus().keySet().toArray(): + mem_status = sc._jsc.sc().getExecutorMemoryStatus().get(executor) + total = mem_status._1() / (1024**3) + free = mem_status._2() / (1024**3) + print(f"{executor}: {total:.2f}GB total, {free:.2f}GB free") +``` + +### Pattern 5: Shuffle Optimization（Shuffle 优化） + +```python +# Reduce shuffle data size +spark.conf.set("spark.sql.shuffle.partitions", "auto") # With AQE +spark.conf.set("spark.shuffle.compress", "true") +spark.conf.set("spark.shuffle.spill.compress", "true") + +# Pre-aggregate before shuffle +df_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 distinct +distinct_count = df.select("category").distinct().count() + +# GOOD: Approximate distinct (no shuffle) +approx_count = df.select(F.approx_count_distinct("category")).collect()[0][0] + +# Use coalesce instead of repartition when reducing partitions +df_reduced = df.coalesce(10) # No shuffle + +# Optimize shuffle with compression +spark.conf.set("spark.io.compression.codec", "lz4") # Fast compression +``` + +### Pattern 6: Data Format Optimization（数据格式优化） + +```python +# 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 columns +df = (spark.read.parquet("s3://bucket/data/") + .select("id", "amount", "date")) # Spark only reads these columns + +# Predicate pushdown - filter at storage level +df = (spark.read.parquet("s3://bucket/partitioned/year=2024/") + .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 queries +spark.sql(""" + OPTIMIZE delta.`s3://bucket/delta_table/` + ZORDER BY (customer_id, date) +""") +``` + +### Pattern 7: Monitoring and Debugging（监控与调试） + +```python +# Enable detailed metrics +spark.conf.set("spark.sql.codegen.wholeStage", "true") +spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true") + +# Explain query plan +df.explain(mode="extended") +# Modes: simple, extended, codegen, cost, formatted + +# Get physical plan statistics +df.explain(mode="cost") + +# Monitor task metrics +def analyze_stage_metrics(spark): + """Analyze recent stage metrics""" + status_tracker = spark.sparkContext.statusTracker() + + for stage_id in status_tracker.getActiveStageIds(): + stage_info = status_tracker.getStageInfo(stage_id) + print(f"Stage {stage_id}:") + print(f" Tasks: {stage_info.numTasks}") + print(f" Completed: {stage_info.numCompletedTasks}") + print(f" Failed: {stage_info.numFailedTasks}") + +# Identify data skew +def check_partition_skew(df): + """Check for partition skew""" + partition_counts = (df + .withColumn("partition_id", F.spark_partition_id()) + .groupBy("partition_id") + .count() + .orderBy(F.desc("count"))) + + partition_counts.show(20) + + stats = partition_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_ratio = stats["max"] / stats["avg"] + print(f"Skew ratio: {skew_ratio:.2f}x (>2x indicates skew)") +``` + +## Configuration Cheat Sheet（配置速查表） + +```python +# Production configuration template +spark_configs = { + # Adaptive Query Execution (AQE) + "spark.sql.adaptive.enabled": "true", + "spark.sql.adaptive.coalescePartitions.enabled": "true", + "spark.sql.adaptive.skewJoin.enabled": "true", + + # Memory + "spark.executor.memory": "8g", + "spark.executor.memoryOverhead": "2g", + "spark.memory.fraction": "0.6", + "spark.memory.storageFraction": "0.5", + + # Parallelism + "spark.sql.shuffle.partitions": "200", + "spark.default.parallelism": "200", + + # Serialization + "spark.serializer": "org.apache.spark.serializer.KryoSerializer", + "spark.sql.execution.arrow.pyspark.enabled": "true", + + # Compression + "spark.io.compression.codec": "lz4", + "spark.shuffle.compress": "true", + + # Broadcast + "spark.sql.autoBroadcastJoinThreshold": "50MB", + + # File handling + "spark.sql.files.maxPartitionBytes": "128MB", + "spark.sql.files.openCostInBytes": "4MB", +} +```