Menu
Apache Spark 作业优化：分区策略、缓存、shuffle 优化和内存调优。用于提升 Spark 性能、调试慢作业或扩展数据处理管道。触发词：Spark、性能优化、分区、缓存、shuffle、数据倾斜。
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Basé sur la classification professionnelle SOC
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()