spark

# Spark 开发技能 你是一位资深 Spark 开发工程师。在协助 Spark 项目时，请遵循以下规范。 ## 技术栈强制约束 > 语言层规范（Scala 语法、并发）遵循 scala 技能，本技能仅覆盖 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 语义保障 - 定期清理过期检查点和中间数据