Menu
Apache Spark 作业优化：分区策略、缓存、shuffle 优化和内存调优。用于提升 Spark 性能、调试慢作业或扩展数据处理管道。触发词：Spark、性能优化、分区、缓存、shuffle、数据倾斜。
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Based on SOC occupation classification
LLM 微调专家。使用场景：微调 LLMs、训练自定义模型、适配基础模型进行特定任务。用于配置 LoRA/QLoRA 适配器、准备 JSONL 训练数据集、设置超参数、适配器训练、迁移学习、Hugging Face PEFT 微调、指令微调、RLHF、DPO 或量化部署微调模型。触发词：LoRA、QLoRA、PEFT、微调、适配器调优、LLM 训练、模型训练、自定义模型。
深度研究工具。仅在用户明确说'深度研究'、'深入调查'、'全面报告'时使用。比普通搜索慢 10-100 倍。对于普通研究/查找请求，使用 parallel-web-search。支持多轮：传递 --previous-interaction-id 以继续之前的上下文。
AI 模型提示工程指南：LLM、图像生成器、视频模型。技术：思维链、少样本、系统提示、负提示。模型：Claude、GPT-4、Gemini、FLUX、Veo、Stable Diffusion 提示。用于：更好的 AI 输出、一致的结果、复杂任务、优化。触发词：提示工程、如何提示、更好的提示、提示技巧、提示指南、LLM 提示、图像提示、AI 提示、提示优化、提示模板、提示结构、有效提示、提示技术。
RAG（检索增强生成）模式专家。用于构建 RAG 管道、嵌入文档、实现混合搜索、上下文检索、HyDE、Agentic RAG、多模态 RAG、查询分解、重排或 pgvector 搜索。触发词：RAG、检索、LLM、上下文、嵌入、HyDE、重排、pgvector、多模态。
Web 研究工具：搜索多个来源、综合发现并生成带引用的研究报告。用于用户要求在线研究主题、搜索网页、查找信息、获取最新信息、对比选项或生成研究报告。触发词：研究、网上搜索、查找信息、最新信息、对比选项、研究报告。
帮助用户发现和安装 agent skills。当用户询问"如何做 X"、"找一个做 X 的 skill"、"有没有能...的 skill"，或表达想扩展能力时触发此 skill。
| name | spark-optimization |
| description | Apache Spark 作业优化：分区策略、缓存、shuffle 优化和内存调优。用于提升 Spark 性能、调试慢作业或扩展数据处理管道。触发词：Spark、性能优化、分区、缓存、shuffle、数据倾斜。 |
生产级 Apache Spark 作业优化模式，包括分区策略、内存管理、shuffle 优化和性能调优。
Driver Program
↓
Job（由 action 触发）
↓
Stages（由 shuffle 分隔）
↓
Tasks（每个分区一个）
| 因素 | 影响 | 解决方案 |
|---|---|---|
| Shuffle | 网络 I/O、磁盘 I/O | 最小化宽转换 |
| 数据倾斜 | 任务时长不均 | 加盐、广播连接 |
| 序列化 | CPU 开销 | 使用 Kryo、列式格式 |
| 内存 | GC 压力、数据溢出 | 调优执行器内存 |
| 分区 | 并行度 | 合理设置分区大小 |
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
# 创建优化的 Spark session
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/")
# 计算最优分区数
def calculate_partitions(data_size_gb: float, partition_size_mb: int = 128) -> int:
"""最优分区大小：128MB - 256MB"""
return max(int(data_size_gb * 1024 / partition_size_mb), 1)
# 重新分区以实现均匀分布
df_repartitioned = df.repartition(200, "partition_key")
# 合并以减少分区（无 shuffle）
df_coalesced = df.coalesce(100)
# 1. 广播连接 - 小表连接
# 最佳条件：一侧 < 10MB
small_df = spark.read.parquet("s3://bucket/small_table/") # < 10MB
large_df = spark.read.parquet("s3://bucket/large_table/") # TBs
# 显式广播提示
result = large_df.join(
F.broadcast(small_df),
on="key",
how="left"
)
# 2. 排序合并连接 - 大表的默认选择
result = large_df1.join(large_df2, on="key", how="inner")
# 3. 桶连接 - 预排序，连接时无 shuffle
(df.write
.bucketBy(200, "customer_id")
.sortBy("customer_id")
.mode("overwrite")
.saveAsTable("bucketed_orders"))
# 4. 倾斜连接处理
# 启用 AQE 倾斜连接优化
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")
from pyspark import StorageLevel
# 缓存重用 DataFrame
df = spark.read.parquet("s3://bucket/data/")
df_filtered = df.filter(F.col("status") == "active")
# 在内存中缓存（默认 MEMORY_AND_DISK）
df_filtered.cache()
# 强制物化
df_filtered.count()
# 存储级别说明：
# MEMORY_ONLY - 快，但可能放不下
# MEMORY_AND_DISK - 溢出到磁盘（推荐）
# MEMORY_ONLY_SER - 序列化，占用内存少，CPU 多
# DISK_ONLY - 内存紧张时
# 执行器内存配置
# spark-submit --executor-memory 8g --executor-cores 4
# 内存分解（8GB 执行器）：
# - spark.memory.fraction = 0.6（60% = 4.8GB 用于执行 + 存储）
# - spark.memory.storageFraction = 0.5
# - 40% = 3.2GB 用于用户数据结构和内部元数据
spark = (SparkSession.builder
.config("spark.executor.memory", "8g")
.config("spark.executor.memoryOverhead", "2g")
.config("spark.memory.fraction", "0.6")
.config("spark.memory.storageFraction", "0.5")
.getOrCreate())
# 减少 shuffle 数据大小
spark.conf.set("spark.sql.shuffle.partitions", "auto") # 配合 AQE
spark.conf.set("spark.shuffle.compress", "true")
# Shuffle 前预聚合
df_optimized = (df
.groupBy("key", "partition_col")
.agg(F.sum("value").alias("partial_sum"))
.groupBy("key")
.agg(F.sum("partial_sum").alias("total")))
# Parquet 优化
(df.write
.option("compression", "snappy")
.option("parquet.block.size", 128 * 1024 * 1024)
.parquet("s3://bucket/output/"))
# Delta Lake 优化
(df.write
.format("delta")
.option("optimizeWrite", "true")
.option("autoCompact", "true")
.mode("overwrite")
.save("s3://bucket/delta_table/"))
# 启用详细指标
spark.conf.set("spark.sql.codegen.wholeStage", "true")
# 解释查询计划
df.explain(mode="extended")
# 识别数据倾斜
def check_partition_skew(df):
partition_counts = (df
.withColumn("partition_id", F.spark_partition_id())
.groupBy("partition_id")
.count()
.orderBy(F.desc("count")))
skew_ratio = stats["max"] / stats["avg"]
print(f"Skew ratio: {skew_ratio:.2f}x (>2x indicates skew)")
# 生产配置模板
spark_configs = {
# 自适应查询执行（AQE）
"spark.sql.adaptive.enabled": "true",
"spark.sql.adaptive.coalescePartitions.enabled": "true",
"spark.sql.adaptive.skewJoin.enabled": "true",
# 内存
"spark.executor.memory": "8g",
"spark.memory.fraction": "0.6",
"spark.memory.storageFraction": "0.5",
# 并行度
"spark.sql.shuffle.partitions": "200",
# 序列化
"spark.serializer": "org.apache.spark.serializer.KryoSerializer",
# 广播
"spark.sql.autoBroadcastJoinThreshold": "50MB",
}
.count() 判断存在 - 使用 .take(1) 或 .isEmpty()