上百T数据清洗过滤：Spark性能调优与架构实战

当数据量到了“上百T”这个级别清洗过滤就不再是写几个过滤条件那么简单了。我接到过不少这样的需求几十台机器堆在那里任务跑了一晚上第二天一看要么OOM要么整个集群被拖垮要么产出的数据质量压根没法看。Spark 确实是为大数据而生的但“能跑 Spark”和“能跑好上百T的清洗过滤”是两码事。这篇文章我想聊聊我在处理这个量级数据时踩过的坑、总结出的思路以及一套可以照着搭的清洗过滤框架。收到这个题目时我的第一反应是这哪是问技术选型分明是问“你怎么在不炸集群的前提下把活干完”。上百T的数据单机存都存不下连遍历一次都要按小时算更别提中间还要清洗、过滤、去重、关联。但 Spark 的分布式模型决定了它天生适合这种场景——前提是你真正理解它的分区、Shuffle、内存模型并且愿意为了数据倾斜去设计过滤顺序。这篇文章适合那些刚接手大数据平台的工程师也适合已经在跑但一直性能不理想的团队我会把从读入到写出的完整链路、性能调优的关键参数以及我实际遇到过的坑全部摊开来讲。1. 为什么上百T的清洗过滤会让很多团队头疼1.1 数据规模带来的三大变化当你把一个数据集的规模从 G 级提升到 T 级甚至上百T很多在中小数据上习以为常的操作会彻底失效。首先是全量扫描的成本暴涨。一张 100T 的表哪怕用最快的 NVMe 磁盘全表读一遍也要几十分钟而清洗过滤往往不是读一遍就能完事的中间可能还要 Join 维表、去重、重新分区每次操作都是一次或多次全量落地。其次是单点故障的概率被放大。上百T的数据分布在几百上千台节点上任何一台机器的硬件问题、网络抖动、磁盘故障都可能导致整个 Stage 重算而 Spark 的容错机制是依赖血缘关系的重算一个 100T 的 Stage 不是闹着玩的。最后是资源管理的复杂度。你的 Executor 内存、CPU 核数、并行度、分区数量任何一个设置不对都可能让任务从“能跑”变成“完全跑不动”。1.2 清洗过滤的本质不是“过滤”而是“控制数据流”我在处理这种量级数据时最大的一个认知转变是清洗过滤的难点不在于写过滤条件而在于控制数据流的方向。你写一个filter(df.col(status) active)很简单但如果这个过滤之后你还想按日期重新分区那问题就来了——数据会从几百个原始分区打散重组这中间的 Shuffle 代价比过滤本身高得多。所以做上百T的清洗你必须想清楚每一条数据的行进路径从哪里读进来在哪一步过滤在哪一步落地哪些操作是宽依赖Shuffle哪些是窄依赖可以在内存里完成。只有把整个数据流画清楚你才知道瓶颈在哪里。2. Spark 的分布式计算模型凭什么能扛住上百T2.1 RDD/DataFrame 的内存计算逻辑Spark 之所以能在上百T数据上做清洗过滤核心靠的是内存计算和懒执行。你可以把整个清洗流程定义成一张 DAG 图Spark 会等你知道最后要写出什么结果时才开始真正计算。这带来一个巨大的优势它可以做谓词下推把过滤条件尽量推到数据源层面比如从 Parquet 文件读数据时直接跳过不符合条件的行组和列这在上百T的场景里能帮你省掉三分之一甚至更多的 I/O。DataFrame 和 Dataset 相比 RDD 更有优势因为 Catalyst 优化器会做自动的谓词下推、常量折叠、列裁剪。我在实际项目中几乎只写 DataFrame 操作很少碰 RDD 裸 API。2.2 分区与并行度让每台机器只干一小块活很多人对分区的理解停留在“分区多了就快”实际上不是。Spark 的一个分区对应一个计算任务分区数量小于 Executor 数量时部分核心会空闲分区数量远大于 Executor 数量时任务调度的开销会吞掉性能。处理上百T数据时我一般会先估算数据量的物理分布然后根据集群总核数来设并行度。举个例子如果你有 100 个 Executor每个 8 核那并行度建议在 800 到 1600 之间而不是直接跑到 10000。分区数决定 Shuffle 的粒度也决定最后写出文件的数量。我后面会详细讲怎么用repartition和coalesce来控制分区数这里先记住一个原则分区的目标不是让 Spark 忙碌而是让每个任务处理的数据量在合理范围内如果太小小于 100MB就会产生大量小任务太大超过 1GB则容易内存溢出或长尾。3. 我常用的清洗过滤架构从读入到写出3.1 数据读入与 Schema 定义上百T数据的源头可能是 Parquet、ORC、Avro也可能是 CSV 这种纯文本。我强烈建议你在源端尽量使用列式存储格式因为清洗过滤基本只关心一小部分列列式存储能大大减少读取的数据量。读入时不要偷懒用spark.read.load(path)自动推断手动定义 Schema 很重要。原因有两点一是自动推断会额外多读一次数据去猜类型在上百T数据上等于白跑一遍二是脏数据可能会导致推断出来的类型完全错误比如某一个字符串列混进了非数字值Schema 就被推断成字符串后续所有过滤都白搭。我的做法是在上线前先用小样本数据跑一次推断然后把 Schema 固化在代码或配置里线上任务直接用固定的 Schema 读入。Python 代码示例from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, LongType, TimestampType spark SparkSession.builder.appName(clean_job).config(spark.sql.shuffle.partitions, 1600).getOrCreate() schema StructType([ StructField(user_id, StringType(), True), StructField(action, StringType(), True), StructField(ts, TimestampType(), True), StructField(event_id, LongType(), True) ]) df spark.read.schema(schema).parquet(hdfs://data/raw/events/)3.2 过滤规则的拆解与编排清洗过滤的逻辑从来不是单一的常见的操作有去重、过滤空值、过滤异常字段、按主键去重、与维度表关联排除某些用户等。我会先把这些规则排一个优先级原则是先做按列过滤再做行级去重最后做需要 Shuffle 的关联过滤。原因很简单按列过滤是窄依赖每个分区的数据独立处理不产生 Shuffle行级去重如果用一个或几个字段做主键即使需要 Shuffle也只在那个小字段上进行关联过滤则是最重的操作最好前面已经过滤掉大量脏数据再执行。比如先filter(df[action].isNotNull())能去掉 30% 的行再去重和关联整个 Shuffle 数据量就小很多。3.3 写出策略Parquet 分区与压缩清洗过滤的结果写到哪里怎么写往往比过滤本身更影响下一次使用。我的默认选择是 Parquet Snappy 压缩并按某个业务字段如日期做分区。这里的“分区”和 Spark 的内存分区是两个概念物理分区能让后续查询直接只读相关目录大幅减少 I/O。写出的精简配置如下clean_df df.filter(df[action].isNotNull())\ .dropDuplicates([user_id, event_id]) clean_df.write.mode(overwrite)\ .partitionBy(dt)\ .option(compression, snappy)\ .parquet(hdfs://data/clean/events/)这里要注意写出时 Spark 默认每个内存分区都会生成一个文件如果你并行度是 2000那就会产生 2000 个小文件。文件太多对下游的读取会造成巨大压力。我通常会在写出前coalesce或repartition到一个可控的文件数量比如按分区目录的块大小估算让每个文件大约 256MB 到 512MB。4. 性能与资源调优跑通不等于跑快4.1 并行度、分区数与 CPU/内存的匹配调优的第一步是搞清楚你的集群有多少资源。假设你有 10 个节点每个节点 32 核、128GB 内存Spark Executor 一般配置为每个节点 4 个 Executor每个 8 核 32GB留一部分内存给 OS。那么总 Executor 数是 40并行度设置在 320 到 640 合适初步设置spark.sql.shuffle.partitions为 400 左右再监控实际运行情况。上百T的数据清洗每 Task 处理的数据量最好控制在 128MB 到 512MB 之间这样可以保证 Spark 能一次性把数据放入内存计算减少溢写到磁盘的频率。我初期喜欢用自动调度慢慢试但后来都是直接用固定配置加监控因为上百T的任务跑一次太贵不能用试错法。4.2 数据倾斜的检测与处理数据倾斜是上百T场景里最典型的杀手。比如你按user_id去重但超级用户有几十亿条记录那这个 key 所在的分区就会拖死整个 Stage。检测倾斜最简单的办法是看 Spark UI 里 Stage 运行时间柱状图如果有个别 Task 显著慢于其他 Task基本就是倾斜了。处理倾斜的方法很多最常用的是加盐。过滤去重场景下我试过把user_id拼接随机数字进行二次聚合然后去掉盐再聚合。另一种思路是过滤之前就把极端倾斜的 key比如已知的爬虫、内测用户提出来单独处理剩下的数据自然均匀。对于上百T的项目我强烈建议把倾斜 key 当作数据质量问题去治理而不是在 SQL 里打补丁。4.3 Shuffle 的代价与避免Shuffle 是 Spark 最昂贵的操作它会将数据跨节点传输并可能写到磁盘。做上百T数据的清洗哪怕只有一次groupBy或distinct操作Shuffle 的数据量都可能达到几十T。我通常的策略是尽量减少 Shuffle 次数能在一个 Stage 完成的过滤尽可能合并条件。其次调整spark.shuffle.partitions能影响 Shuffle 输出的小文件数量过大会导致每次最后一个 Task 拉取大量文件过小则单个 Partition 数据量太大。此外利用broadcast把小维度表广播到每个 Executor 上可以把一个极其耗时的 Shuffle Join 变成 Map 端的简单过滤这个技巧在过滤上游用户名单、黑名单时特别有效。5. 踩过的坑与绕过方法5.1 OOM 问题不止是加大内存一个 100T 的清洗任务我最初以为 JVM 堆内存不够就加大 Executor 内存。结果 Executor 直接变成 64GB任务照样挂而且 GC 卡顿更严重。后来才明白OOM 很多时候是 Spark 的 Tungsten 管理机制和你的分区大小不匹配导致的。Spark 内部缓存数据的结构叫UnsafeRow它是紧凑的内存表示非常省内存但如果你的单分区数据量太大超过了该 Executor 的可用内存就会溢写到磁盘而溢写需要大量序列化和反序列化CPU 瞬间打满。真正的解法是增加分区数降低每个 Task 的数据量同时设置spark.memory.offHeap.size和spark.memory.offHeap.enabled来利用堆外内存减轻 GC 压力。每台机器留出 20% 的退避内存给系统否则你用满后会出现整机崩溃。5.2 小文件灾难清洗后反而更慢这是我吃过最大的亏。清洗之前的数据是几百个大的 Parquet 文件处理完写出后反而变成了几万个几十 KB 的小文件下游跑查询比原来还慢。原因就是我前面说的并行度太高每个 Task 都往目录里写一个小文件。解决方法是洗完的数据在写出前进行coalesce如果你有 400 个分区把它coalesce到 64 个写入后每个文件就大约是 700MB 到 1GB完全可控。我后来还专门写了一个检查任务统计输出目录的文件数和每个文件大小防止这种问题再次发生。5.3 空值、脏数据的隐性风险清洗过滤里最容易被忽略的是空值和脏数据带来的逻辑陷阱。比如null值与任何值比较都返回null在filter里会被当作false是真的但join时null与null可能匹配也可能不匹配行为很反直觉。还有一个坑是 CSV 读入时多余的空格、不可见字符都会导致过滤条件失效。我建议在每个清洗任务上线前先跑一个小规模的采样把字段的null占比、空字符串占比、类别分布展示出来你才能知道规则到底能不能命中。真正跑上百T之前先在 1T 采样上调试等采样数据跑通了再放全量这是节省时间和心情的最好办法。6. 从清洗到数据治理的扩展思考6.1 把过滤逻辑沉淀成可配置的规则上百T的清洗过滤如果每次都是硬编码在 PySpark 脚本里一旦业务改一个条件你就要重新跑几十个小时。我后期的做法是把过滤条件抽象成配置比如 JSON 或者 YAML里面描述字段名、类型、取值范围、空值策略、去重主键然后用一个通用的 Spark 程序解析配置并执行。这样新增一个清洗任务只需要改配置不需要动代码。而且配置能被做成一个规则中心后续可以直接映射到数据治理平台。这个思路适合清洗任务特别多、每天都要跑的团队成本是开发一套规则解析器初期大概一周工作量但长期节省的时间非常可观。6.2 增量清洗与流式清洗的雏形最后聊一下增量。清洗上百T全量数据在很多场景下是不合理的比如日志数据每天都在增长每天重跑全量既浪费资源又不实时。我后来的方案是把“历史全量增量新数据”分开处理历史数据每周做一次校验和重清洗增量数据用 Streaming 配合Append模式按天实时清洗。Streaming 在处理清洗过滤的原理上和批处理完全一致只是把数据来源变成了 Kafka。关键点是保持过滤规则的同一套配置避免增量处理和批量处理结果不一致。如果你现在还没上全量清洗我建议你一开始就考虑增量架构否则等到全量数据量到几百T再改造会非常痛苦。跑过几次上百T的清洗任务后我最大的感受就是“敬畏数据”。每一张表都像一条大河你以为你只是在河边捞捞浮沫但实际上每一股水流都可能带着暗礁。Spark 给了你一艘大船但掌舵的依然是你——你的分区设置、过滤编排、写出策略决定了这艘船是平稳过弯还是直接搁浅。如果你也在准备处理类似量级的数据记住我刚提到的原则先采样跑通再全量跑先过滤窄依赖再处理宽依赖先关注文件大小再调整并行度。数据和代码都会说谎但监控面板上的运行时间不会。祝顺利。