企业官网建设流程全解析

当数据量到了“上百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 的分布式计算模型：凭什么能扛住上百T

2.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 给了你一艘大船，但掌舵的依然是你——你的分区设置、过滤编排、写出策略，决定了这艘船是平稳过弯还是直接搁浅。如果你也在准备处理类似量级的数据，记住我刚提到的原则：先采样跑通，再全量跑；先过滤窄依赖，再处理宽依赖；先关注文件大小，再调整并行度。数据和代码都会说谎，但监控面板上的运行时间不会。祝顺利。