300w数据集高效处理实战：从ETL到分布式计算的架构演进

共计 1607 个字符，预计需要花费 5 分钟才能阅读完成。

背景痛点

在处理 300 万级别的数据集时，单机环境往往会遇到以下几个典型问题：

1. 内存限制：JVM 堆内存通常设置为 4 -8GB，加载 300w 条结构化数据（假设每条 1KB）至少需要 3GB 内存，加上中间计算对象，极易触发 OOM（Out of Memory）
2. GC 停顿：频繁的 Full GC 会导致处理线程暂停，实测单机处理 300w 条 JSON 数据时，GC 时间占比可达 15%-20%
3. 计算效率：单线程处理耗时呈线性增长，实测 i7 处理器单线程处理 300w 条 CSV 数据需要约 25 分钟

技术选型

Spark vs Flink 对比

通过 TPCx-BB 基准测试（100GB 数据集），得到如下实测数据：

选型建议：
– 需要低延迟选 Flink
– 需要更好生态整合选 Spark
– 中等规模数据 (300w-500w) 两者差异不大

核心实现

分区策略黄金法则

1. 初始分区数：建议取集群总核数的 2 - 3 倍

   # 根据集群配置自动计算
   optimal_partitions = sc.defaultParallelism * 2

2. coalesce vs repartition：
3. coalesce：合并分区无 shuffle，适合过滤后减少分区
4. repartition：触发 shuffle，适合均匀分布数据

PySpark 优化示例

from pyspark.sql import SparkSession

# 初始化配置（显示设置执行器内存）spark = SparkSession.builder \
    .appName("300w_etl") \
    .config("spark.executor.memory", "4g") \
    .getOrCreate()

# 最佳实践：使用 parquet 格式 + 谓词下推
(df.read.parquet("input.parquet")
   .filter("amount > 100")  # 先过滤再处理
   .repartition(optimal_partitions)  # 显式重分区
   .createOrReplaceTempView("transactions"))

# 使用 SQL 进行聚合（自动优化执行计划）spark.sql("""
  SELECT user_id, SUM(amount) 
  FROM transactions 
  GROUP BY user_id
""").show()

性能优化

内存配置公式

executor_memory = (数据集大小 * 3) / (并行度 * 0.7)

数据倾斜解决方案

1. 加盐法（Salting）：

   from pyspark.sql.functions import concat, lit, rand
   
   skewed_df = df.withColumn(
     "salted_key", 
     concat(df["user_id"], lit("_"), (rand() * 10).cast("int"))
   )

2. 两阶段聚合：先对倾斜 key 局部聚合，再全局聚合
3. 广播小表：对 join 操作中的小表使用广播变量

避坑指南

OOM 检查清单

• [] 确认 spark.executor.memoryOverhead 设置合理（建议总内存的 10%-15%）
• [] 检查是否存在 collect()操作带回大量数据到 Driver
• [] 监控 Storage Memory 占比（过高说明缓存策略不当）

数据一致性保障

1. 写一致性：

   df.write \
     .mode("overwrite") \
     .option("maxRecordsPerFile", 100000) \
     .parquet("output_dir")

2. 读一致性：启用 HDFS 的 WAL（Write-Ahead Log）

架构演进思考

当数据量增长到 1000w 时，需要考虑：
1. 是否引入增量处理（Delta Lake/Iceberg）
2. 计算存储分离架构（S3+Spark）
3. 细粒度资源调度（K8s+YARN）

欢迎在评论区分享你的架构演进方案！