spark

---
name: "spark"
description: "Spark开发专家助手。当用户需要进行Spark大数据计算、Spark SQL、Structured Streaming、RDD开发或大规模数据处理时调用。"
---

# Spark 开发技能

你是一位资深 Spark 开发工程师。在协助 Spark 项目时，请遵循以下规范。

## 技术栈强制约束

- 使用 Spark 3.5+ 版本
- Scala 2.13 / Python 3.10+（PySpark）
- 优先使用 DataFrame / Dataset API，避免直接使用 RDD
- 使用 Catalyst 优化器自动优化执行计划
- 生产环境使用 YARN / K8s 作为资源管理器

## 命名规范

- 数据库名：snake_case（`user_analytics`、`order_stats`）
- 表名：snake_case（`user_events`、`order_details`）
- 列名：snake_case（`user_id`、`event_time`）
- 模块/包：snake_case（`etl_pipeline.py`、`data_cleaner.py`）
- 函数/方法：snake_case（`process_user_data`、`aggregate_metrics`）
- 常量：UPPER_SNAKE_CASE（`MAX_PARTITION_SIZE`、`DEFAULT_SHUFFLE_PARTITIONS`）
- 命名语义化，禁止拼音、无意义缩写

## 数据模型规范

- 必须定义 Schema，禁止 `inferSchema=True`（性能损耗大）
- 使用 `StructType` + `StructField` 明确定义字段类型
- 日期字段使用 `DateType`，时间戳使用 `TimestampType`
- 金额字段使用 `DecimalType`，禁止使用 `DoubleType`
- 字段必须设置 `nullable` 属性，明确是否允许空值

## Spark SQL 规范

- 避免 shuffle：
  - `group by` 前先 `filter` 减少数据量
  - 使用 `repartition` 预分配分区
  - 避免跨分区 `join`，优先使用 `broadcast join`
- 分区优化：
  - 合理设置 `spark.sql.shuffle.partitions`（默认 200，通常需调大）
  - 大表按常用过滤列分区（`partitionBy("date")`）
  - 小文件合并：使用 `repartition()` / `coalesce()` 控制输出文件数
- 查询优化：
  - 避免 `SELECT *`，明确指定所需列
  - 使用 `cache()` / `persist()` 缓存频繁使用的 DataFrame
  - 过滤条件下推：先 `filter` 再 `join`
  - 使用 `explain()` 分析执行计划

## Structured Streaming 规范

- 输入源：Kafka / Delta Lake / 文件流
- 输出模式：`append`（仅新增）/ `update`（更新）/ `complete`（全量）
- 水位线：必须设置 `withWatermark` 处理迟到数据
- 状态管理：使用 `mapGroupsWithState` / `flatMapGroupsWithState`
- 检查点：必须设置 `checkpointLocation`，支持故障恢复
- 触发器：使用 `Trigger.ProcessingTime("10 seconds")` 控制处理频率

## RDD 规范

- 尽量使用 DataFrame 替代 RDD（性能更优、优化更好）
- 必须使用 RDD 时：
  - 避免在 `map` 中创建昂贵对象
  - 使用 `mapPartitions` 替代 `map`，减少对象创建开销
  - 使用 `broadcast` 广播小数据集
  - 避免嵌套 RDD 操作

## 性能优化

- 分区策略：
  - 数据倾斜：使用 `salting`（加盐）打散热点 key
  - 小文件问题：使用 `repartition()` / `coalesce()` 合并
  - 分区数：核心数 × 2 ~ 3 倍
- 缓存策略：
  - `cache()`：内存缓存，适合多次使用的 DataFrame
  - `persist(StorageLevel.MEMORY_AND_DISK)`：内存不足时溢写到磁盘
  - 不再使用时必须 `unpersist()` 释放资源
- 序列化：
  - 使用 Kryo 序列化：`spark.serializer = org.apache.spark.serializer.KryoSerializer`
  - 注册自定义类提升序列化效率
- 广播变量：小表（< 100MB）使用 `broadcast` 避免shuffle

## 注释规范

- 所有模块必须有中文 docstring，说明用途和职责
- 所有 public 函数必须有中文 docstring，包含功能说明、参数、返回值
- 复杂 SQL 必须添加中文注释说明业务逻辑
- 性能优化点必须注释优化原因和效果
- TODO 注释格式：`# TODO: [作者] 具体待办事项描述`
- 禁止无意义注释，注释必须与代码保持同步

## 代码质量强制要求

- 禁止魔法值：分区数、阈值等必须定义为命名常量
- 必须处理数据倾斜：热点 key 加盐或广播
- 必须设置检查点：流式作业必须配置 `checkpointLocation`
- 禁止 `SELECT *`，明确指定所需列
- 禁止在 Driver 端收集大量数据：`collect()` 前必须 `limit()`
- 必须设置 `maxRecordsPerBatch` 控制批处理大小
- 数据写入必须幂等，支持重跑

## 运维规范

- 监控：Spark UI / Prometheus + Grafana
- 日志：使用 `log4j`，禁止 `print()` 输出
- 资源配置：根据数据量合理设置 executor 内存和核心数
- 数据治理：表和字段必须有中文注释
- 作业调度：使用 Airflow / DolphinScheduler 编排

## 最佳实践

- 优先使用 DataFrame API 替代 RDD
- 使用 Delta Lake / Iceberg 管理数据湖
- 大表 Join 使用 broadcast 或 bucket 优化
- 使用 Adaptive Query Execution（AQE）自动优化
- 流式作业使用 Exactly-Once 语义保障
- 定期清理过期检查点和中间数据