Google PDE: Spark、Dataproc 与分布式数据处理 — 学习指南

    --conf mapreduce.fileoutputcommitter.algorithm.version=2
    ```

  - 使用 Parquet/ORC 格式，并结合列裁剪和谓词下推。通过合并（compaction）来管理小文件，目标是使每个文件大小在 128–512 MiB 之间，以实现高效扫描。
- Hive 元存储
  - 将模式和表元数据集中到 Dataproc Metastore（托管的 Apache Hive Metastore）或由 Cloud SQL 支持的元存储中，以便在集群间共享目录。
  - 使用指向 GCS 的外部表以确保持久性；按日期/小时进行分区以限制扫描成本。
- 作业、初始化和工作流
  - 提交 `spark`、`pyspark`、`spark-sql` 或 `hadoop` 作业。初始化操作在集群创建时安装额外的库或代理（例如，连接器、Python 库）。
  - 工作流模板可将多步骤流水线参数化；它们可以为每个工作流创建临时集群，然后在完成后将其拆除。这可以改善隔离性并削减空闲成本。
  - 对于批量 ETL，推荐使用临时集群；数据和元存储存在于集群之外（GCS、Dataproc Metastore、BigQuery）。
- BigQuery 集成
  - Spark BigQuery 连接器可直接读写 BigQuery；考虑使用 BigQuery Storage Read API 以获得高吞吐量，使用 Write API 以实现更低延迟、仅一次的流式插入。
  - 对于表维护，在 BigQuery 中执行下游的 MERGE/分区覆盖操作，以原子方式完成加载。
### Spark 模型、性能调优与可靠性

- API 与执行
  - RDDs：低级别、不可变、在 Scala/Java 中是类型安全的；由您控制分区和持久化。
  - DataFrames/Datasets：关系型，经 Catalyst 优化；因其查询优化和代码生成能力，在 ETL 场景中应优先使用。
  - 转换操作（Transformation）是惰性的（如 map、filter、join）；动作操作（Action）触发执行（如 count、collect、save）。Spark 会构建一个由 shuffle 分割的阶段（stage）组成的 DAG；任务（task）在每个分区上运行。
- 分区与 shuffle
  - 输入分区：确保有足够的分区来利用所有核心；初始可设置为 executor 总核心数的 2-4 倍。通过

    --conf spark.sql.shuffle.partitions= {total_executor_cores * 2 to 3}
    ```

  - 目标是让宽转换（wide transform）后的每个分区大小在 100–256 MiB 左右；分区太小会导致调度器开销，太大则有 executor OOM 的风险。
  - 对于 join、groupBy 和 orderBy 操作，Shuffle 是主要的成本开销。确保 executor 有足够的内存和磁盘空间；对于集群上的重度 shuffle 负载，可考虑使用本地 SSD。
- 数据倾斜与 join 策略
  - 检测数据倾斜（任务运行时长出现长尾、分区大小差异悬殊）。缓解措施：
    - 广播小表以避免 shuffle：

      --conf spark.sql.adaptive.enabled=true
      ```

- 缓存、检查点与血缘
  - 当中间 DataFrame 会被重复使用时，谨慎地缓存热点数据；优先使用 MEMORY_AND_DISK 以避免 OOM。
  - 将长血缘关系的数据 checkpoint 到 GCS 或 HDFS，以限制失败时需要重新计算的范围。
- Executor 与动态分配
  - 合理设置 executor 的大小，以平衡并行度与 GC 开销：
    - 每个 executor 的核心数：对于均衡的 I/O/CPU 任务，设置为 2-5 个；较少的核心数可以减少 GC 暂停时间。
    - 内存开销：为宽 shuffle 设置 `spark.yarn.executor.memoryOverhead`。
    - 在集群上启用动态分配和外部 shuffle 服务，使 executor 数量能随工作负载伸缩：

    MERGE target t USING staging s
    ON t.id = s.id
    WHEN MATCHED THEN UPDATE SET ...
    WHEN NOT MATCHED THEN INSERT (...)
    ```

  - 增量处理：在 `ingestion_date` 分区上使用基于水印的过滤；在 GCS 中维护一个已处理清单，以避免重复处理。
  - 死信处理：当发生解析/验证错误时，将坏记录连同诊断信息分流到隔离路径/表中。若需严格的 schema 强制和内置的 DLQ（死信队列），可考虑使用 Dataflow；若使用 Spark，则需实现记录级别的 try/catch 和一个独立的 sink。
### 安全、可观测性与成本

- 身份与访问权限
  - 在具有最小权限 IAM 的专用服务账号下运行集群和作业。仅授予所需角色，例如：
    - roles/dataproc.worker 授予实例服务账号
    - roles/storage.objectViewer 或 objectAdmin 授予 GCS I/O 路径
    - roles/bigquery.dataEditor 授予目标数据集
  - 对于 Dataproc Serverless，使用按作业的服务账号来限定访问范围。
- 网络隔离与加密
  - 在 VPC 子网中使用私有 IP 集群，通过防火墙限制主节点 UI 的访问，并启用 Private Google Access 以便在没有公共出口的情况下访问 GCS/BigQuery。
  - 将集群放置在 Shared VPC 项目中以实现集中控制。可选择在 Dataproc 上启用 Kerberos 以进行集群内身份验证。
  - 使用 CMEK 进行静态加密：在 GCS 存储桶、永久性磁盘、Dataproc Metastore 和 BigQuery 上配置 CMEK；默认使用 TLS 进行传输中加密。
- 日志、历史记录与指标
  - 启用 Spark 事件日志到 GCS 并部署 History Server：

     --conf spark.sql.shuffle.partitions=600
     --conf spark.sql.adaptive.enabled=true
     ```

   - 对小型维度表使用广播连接，并将长血缘关系的数据检查点到 GCS 以提高稳定性。
   - 理由：正确的分区可以减少数据倾斜和调度器开销；AQE 在运行时能适应数据分布；检查点机制限制了故障后需要重新计算的范围。

7) 强化安全与网络
   - 使用专用服务账号运行集群，仅授予 GCS 路径、metastore 和 BigQuery 数据集所需的角色。在受限子网中创建私有 IP 集群，启用 Private Google Access，并通过防火墙规则限制对 UI 的访问。
   - 理由：最小权限原则和网络隔离减少了攻击面；私有的控制平面出口避免了公网暴露。

8) 配置日志、历史记录和警报
   - 启用 Spark 事件日志到 GCS 并部署 History Server；将驱动程序/YARN 日志路由到 Cloud Logging 并设置保留策略。为长时间待处理的容器、重复的任务失败或过长的作业持续时间添加 Monitoring 警报。
   - 理由：集中式日志支持根本原因分析；主动警报能及早发现数据倾斜、OOM 或 I/O 性能下降等问题。

9) 通过 Dataproc Serverless 对临时和弹性峰值负载进行选择性现代化改造
   - 将零星的或探索性的 Spark SQL 工作负载迁移到 Dataproc Serverless；在无服务器模式上得到充分验证之前，保持夜间管道在临时集群上运行。
   - 理由：无服务器模式消除了集群运维工作并能自动扩缩容，非常适合不可预测的负载；现有工作流可以继续运行，只需极少的代码更改。

10) 验证对象存储提交器和小文件管理
    - 设置 FileOutputCommitter v2 算法，并在写入前通过 repartition/coalesce 将输出文件合并为每个文件 256–512 MiB。
    - 理由：对象存储缺少原子性的重命名操作；优化的提交器减少了复制/重命名开销。文件合并缓解了小文件问题，从而提升性能并降低成本。

此设计以最小的重构代价重用了现有的 Spark 和 Hive 作业，确保了数据在 GCS 中的持久性，集中了模式定义，控制了安全风险范围，提供了强大的可观测性，并通过临时集群、自动扩缩容、可抢占容量以及有针对性地使用无服务器执行来优化成本。

---

← [消息传递、事件注入与实时服务](/cn/posts/pde-messaging-ingestion/)  ·  [所有领域](/cn/posts/google-pde-study-guide/)  ·  [数据注入、集成与迁移](/cn/posts/pde-ingestion-migration/) →

**[练习这些题目 →](/cn/kb/google/)**  ·  **[在 ExamRoll.io 上限时练习 →](https://www.examroll.io/?utm_source=guide&utm_medium=referral&utm_campaign=PDE)**