1. 先搞清楚一件事HDFS 和 Spark 集成到底在解决什么问题很多人一上来就谈参数调优谈内存分配但我觉得有个更根本的问题得先聊透:HDFS 和 Spark 本来就不是很“般配”的一对它们能凑在一起工作本质上是大数据技术栈在特定历史阶段下的一种“务实选择”。我先说说这两兄弟的身世。HDFS 诞生于 Hadoop 生态它擅长的是海量文件的分布式存储——默认块大小 128MB写入一次、读取多次为了容错会把每个块复制三份。但它的计算能力几乎为零它只管存不管算。而 Spark 恰好相反它是一个内存计算框架擅长把复杂的数据处理任务拆成一个个阶段Stage去跑但它自己不带存储系统数据总得有地方放。所以在大数据平台里最常见也最经典的一种架构就是HDFS 负责存Spark 负责算两者通过网络对接。这套组合天然就有几个问题一是网络开销大。Spark 从 HDFS 读数据本质上是把远端节点的数据拉到 Executor 所在节点去处理。如果数据在节点 A而跑任务的 Executor 在节点 B那中间就隔着一次网络传输。集群规模一大节点上百台每次任务都要跨节点拉数据整体吞吐立刻就成了瓶颈。二是元数据压力。HDFS 的 NameNode 是单点它维护着整个文件系统的目录树和文件/块映射信息。Spark 提交一个作业如果读一张有几万个分区的表那它就要向 NameNode 发起海量元数据请求。三是文件大小分布不均。HDFS 不适合存大量小文件而 Spark 的 partition 数量和输入文件数量/大小直接相关文件一多且碎任务开销就成倍增长。我在实际项目中见过太多团队把 Spark 跑得慢归咎于“集群不够大”或“代码写得烂”实际上有相当一部分性能问题出在 HDFS 和 Spark 的衔接层——文件格式选错、压缩方式不对、数据没有做分区裁剪、跑任务的时候数据本地性极差。这些不是靠堆机器能解决的根子在于你没把这两个系统当做一个整体来设计。这篇文章我就围绕“HDFS Spark 集成”这条主线把我实操中积累的性能优化方法拆开揉碎了讲清楚。内容包括读数据的路径优化、文件布局与格式选择、资源调度与本地性配置、Shuffle 阶段的调优以及一套可以直接拿去用的诊断手段。每个部分我都会给到具体的配置参数和验证方法而不是只讲空洞的原理。2. 读数据链路HDFS 到 Spark 的路径上藏着最多性能损耗2.1 一次读数据的完整旅程我们先画一条线Spark 提交一个作业读取 HDFS 上一个 10GB 的文件这条链路上到底发生了什么。第一步Spark Driver 通过 Hadoop 客户端 API 向 NameNode 发起请求获取这个文件对应的 Block 位置列表。NameNode 会返回每个 Block 所在的 DataNode 列表通常是三个副本对应的节点。第二步Driver 根据这些位置信息划分 RDD 分区。默认情况下HDFS 文件被切分成的分区数 文件大小 / 块大小128MB。每个分区对应一个或多个 Block 的连续区间。第三步Executor 启动 TaskTask 内部通过 Hadoop InputFormat 去 DataNode 拉取数据。这里的关键动作叫“数据本地性调度”如果 Task 被分配到了一个 Block 副本所在的节点上那它走的是本地读路径Short-Circuit Local Reads直接读本地磁盘文件速度快到惊人如果 Task 被分配到其他节点那就要走 DataNode 的 RPC 接口把 Block 数据通过网络传过去。第四步数据进入 Spark 的 Memory 或 DiskStore然后被反序列化成 Java 对象进入 RDD 的计算管线。链路不长但每一步都有可优化的空间。我按我踩坑的严重程度排个序。2.2 数据本地性90% 的人没认真查过的配置先说影响最直接的一个数据本地性Data Locality。Spark 的 TaskScheduler 在分发 Task 时会根据 DAGScheduler 提交的 taskset 里标注的 preferredLocations 去匹配 Executor。它会按PROCESS_LOCAL进程内本地→ NODE_LOCAL同节点本地→ RACK_LOCAL同机架→ ANY跨网络的顺序依次尝试。默认的spark.locality.wait是 3 秒也就是说如果数据本地性达不到较优的级别调度器会在该级别等待 3 秒然后再放宽。生产环境里常见的现象是Task 迟迟起不来日志里大量出现 “Locality Level: ANY”跑完之后看监控面板输入数据全是通过远程网络传输的。我遇到过最夸张的一次一个 1TB 的 Hive 表做聚合集群 50 个节点跑了一个小时还没结束。后来查 Spark UI发现几乎所有 Task 的本地性都是 NODE_LOCAL 和 RACK_LOCALPROCESS_LOCAL 的比例不到 5%。原因是什么执行器的内存里缓存了大量旧任务的 RDD新任务的输入数据虽然在同节点但 Executor 没资源了Task 被放到了其他节点。排查方法很简单在 Spark UI 的 Stages 页面点开任意一个 Stage看 “Locality Level Summary” 这个表格。如果发现全是 ANY 或者 RACK_LOCAL 占大头那就是本地性出了问题。常规解法有这么几种增大 Executor 内存避免因为内存不足把缓存刷掉导致重复计算。调整spark.locality.wait比如从 3 改成 5 甚至 10给调度器更多时间去匹配本地节点。别小看这个参数我曾经把一批小时级作业的耗时从 50 分钟压到 28 分钟就靠这一条。检查 input split 的划分方式。HDFS 的块和 Spark 的输入分区不总是一一对应的spark.hadoop.mapreduce.input.fileinputformat.split.maxsize如果被改了会导致跨节点读数据破坏本地性。还有一个容易忽略的点开启 HDFS Short-Circuit Local Reads。这个功能让 DataNode 上的数据读取绕过 TCP 栈直接用本地文件描述符来读。需要配置 DataNode 的dfs.domain.socket.path同时把dfs.client.read.shortcircuit打开。在 CDH 和 HDP 发行版里这是默认开启的但自己用手工方式搭的 Hadoop 集群比如 Apache 纯原版经常是关着的。要验证当前是否开启可以在 DataNode 日志里搜 “shortcircuit”或者跑一个本地读的任务看数据吞吐量是否异常低。2.3 文件格式和压缩同一个文件差 3 到 8 倍的读取速度这是集成层面能拿到最大收益、但改动成本最低的一部分。HDFS 上存的文件格式直接决定了 Spark 读取时要做多少额外工作。常见的有文本文件TextFile、SequenceFile、Parquet、ORC、Avro 这么几种。我的结论非常明确只要不是没办法生产环境一律用 Parquet 或 ORC别用文本文件。为什么因为列式存储有两个属性对 Spark 极有价值。第一个是谓词下推Predicate Pushdown。也就是说查一张有 100 个字段的表过滤条件是where a5如果文件是列式存储Spark 只需要读取 a 字段和行号索引那一列的数据其他 99 列物理上根本不用加载。这对 IO 的节省不是一倍两倍是数量级的。第二个是压缩比。Parquet 配合 Snappy 或 ZSTD 压缩对结构化数据通常能做到存储压缩比在 5:1 以上。HDFS 上存的文件小读的时候网络传输也少整个链路都快。我放个实测数据表格是之前做的一个电商订单分析任务同样一份数据量约 120GB 的订单明细存储格式压缩方式HDFS 占用全量读取耗时Spark 2.4.8 / 20 个 ExecutorSQL 过滤后再聚合耗时文本 CSV无120GB47 分钟39 分钟文本 CSVGzip35GB22 分钟18 分钟ParquetSnappy22GB9 分钟4 分钟ParquetZSTD18GB8 分钟3 分钟注意 Gzip 文本那一栏——虽然文件变小了读取却并不快。因为 Gzip 压缩的文本文件不支持 splitSpark 只能单线程去解压整个文件等于把一个几 GB 的文件变成了一个无法切分的巨型分区。所以文本文件用 Gzip 压缩在 Spark 场景下是反模式我见过不少团队因为这个原因作业跑得特别慢。Parquet 本身是 splittable 的配合压缩不会破坏并发度。ZSTD 压缩比和速度都比 Snappy 好一点如果你的集群是 CDH 6.x 之后或者 Apache Spark 3.x我建议直接上 ZSTD。2.4 小文件问题HDFS 块大小和 Spark 分区数之间的博弈这个问题我单独拿出来说因为它是 HDFS 和 Spark 集成的“经典老坑”。HDFS 默认块大小是 128MBNameNode 存储元数据的压力决定了它不适合存海量小文件。而 Spark 读取 HDFS 上的一个目录时是一个一个文件去切分的每个小文件都会至少生成一个分区。假设你在 HDFS 上存了 10 万个小文件每个文件 100KB总量约 10GB。Spark 读这个目录会生成 10 万个小分区每个分区就处理 100KB 数据。这是个灾难——10 万个 Task 光调度开销就比数据处理本身还高而且会产生 10 万次对 NameNode 的元数据请求NameNode 通常会因为 RPC 队列爆掉而开始卡顿。解决方案分三个层面写入时治理写数据的时候控制文件数量。用 Spark 写 HDFS 时通过coalesce()或repartition()控制输出文件数目标让每个输出文件接近 128MB。定期合并用 Spark 作业对历史小文件目录做一次“Compact”操作按分区读取后重新写回。写回时利用分区键让数据均匀分布到合理的文件数量。动态分区裁剪读取时用 Hive 分区表的方式组织数据让 Spark 通过分区剪枝只扫描需要的分片目录。我之前给一个项目做过一次存储盘点和整合把 60 万个小文件合并到约 8000 个 128MB 以上的 Parquet 文件。做完之后同一个下游 Spark 批处理作业从 2 小时 10 分钟降到 26 分钟NameNode 的心跳稳定性也上来了。这个优化在整个链路里性价比极高。3. 建表与数据组织给 Spark 一个“对胃口”的 HDFS 目录结构3.1 从 Hive 表到 HDFS 目录一张表就是一个业务契约在很多团队里Spark 读 HDFS 数据不是直接指定路径的而是通过 Hive Metastore 里的表对象来读取。这里有个概念需要点透Hive 表在 HDFS 上的存储目录结构基本决定了 Spark SQL 的性能上限。一张 Hive 表在 HDFS 上的物理形态就是一个目录里面放着数据文件。分区表会在这个目录下再按分区字段建子目录比如order_ds20250101/、order_ds20250102/。这一步非常关键——Spark 在跑 SQL 的时候利用分区信息做裁剪直接决定扫描多少数据。我最常见到的错误用法是建表的时候没有做分区设计或者分区字段选错了。分区字段要选“低基数的过滤字段”比如日期、地区、业务线。不要选用户 ID、订单号这种高基数字段那会导致每个分区数据量极小而分区数量爆炸。分区粒度也要想清楚。你按天分区还是按小时分区如果按小时分区一年的数据就是 8760 个分区目录。对于一张日增 200GB 的表按小时分区还勉强能接受但如果日增只有 20GB一个小时分区只有不到 1GB 数据那分区裁剪的意义就变小了反而增加了元数据开销。我的经验值是单个分区内的数据量至少要在 5GB 以上最好在 50GB 到 500GB 之间。3.2 分区、分桶与排序让数据分布匹配查询模式分桶Bucket是比分区更细粒度的文件组织方式。在 Spark 3.0 之后分桶表配合bucket pruning能实现类似“预聚合”的效果——你不知道具体在哪个分区但通过分桶字段的哈希值能直接定位到对应的桶文件去读。这在相关性 Join比如大表关联小表场景下能显著减少 Shuffle 数据量。但说实话我给你一句真实的建议如果你不是有明确的高频 Join 场景不要轻易引入分桶。分桶表写入逻辑复杂对并发写不友好配合不好反而得不偿失。更实用的技巧是布局排序在写 Parquet 文件之前先按常用的过滤字段比如订单日期、商品类目做一次全局排序。排序之后的数据在写入 Parquet 时min/max 统计信息才能充分发挥作用——Parquet 每个 RowGroup 都会记录该列的最小值和最大值查询时如果过滤条件命中不到 RowGroup 的范围整个 RowGroup 直接跳过。不排序的情况下每个 RowGroup 的 min/max 范围都很大统计信息基本等于废的。我曾经做过一个线上 A/B 对比同一张 300GB 订单表数据文件一个做了全局排序再写 Parquet一个直接写。跑同样的“按用户 ID 查最近 100 条订单”查询排序后的版本扫描的数据量只有未排序版本的 1/5。因为用户 ID 在某一个时间范围内往往集中在某些文件里排序让这些数据物理上靠在一起了。所以我把这个环节总结成一个闭环先用 Hive 分区把时间维度卡死再用 Parquet 列式存储降低列扫描 IO最后在写入前按常用过滤字段排序激活 min/max 裁剪。这三板斧下来绝大多数查询性能都能有个量级上的提升。3.3 存储策略冷热分离和分层存储到了集群运维层面还有个容易被忽视的点HDFS 上的数据不是永远“热”的。Spark 跑批处理既要读最近一天的数据也可能要读三个月前、一年前的数据。如果你把所有数据都放在内存计算集群默认读取的存储策略下集群磁盘 IO 和网络带宽会被“冷数据查询”持续拖累。这里有一个成熟的做法利用 HDFS 的Storage Policy存储策略做冷热分层。比如把近 7 天的数据放到ALL_SSD策略下把 7 天到 90 天的数据放到HOT默认三副本 HDD策略下超过 90 天的数据放到COLD比如归档节点或冷备存储策略下。Spark 读数据时因为 DataNode 会根据策略调度数据到不同介质上热的查询走 SSD冷的查询走 HDD互不干扰。我建议运维同学在生产环境把这套策略落实下去。虽然这是偏向平台侧的优化影响却是实打实的——跑批任务高峰期的平均 IO 延迟能下降 20% 到 40%。4. 资源调度与动态分配让 Executor 和 DataNode 真正“对齐”4.1 Executor 数量、内存分配和并行度的三角平衡聊 HDFS 与 Spark 集成资源调度是逃不开的一环。因为 Spark 的 Executor 数量和每个 Executor 上跑的 Task 并行度直接决定了它到底能同时向 HDFS 发起多少个并发读取流。并行度不够哪怕 HDFS 底层盘阵性能再好Spark 也只能把吞吐打一半并行度太高DataNode 的磁盘 IO 和网络带宽先被打爆NameNode 的 RPC 请求会堆积。我的实践策略是先算并发数再定资源量。假设一个批处理任务要读 1TB 的 Parquet 数据HDFS 平均单节点磁盘吞吐按 200MB/s 估算机械盘阵列SSD 会更好如果希望 15 分钟内完成读取需要的吞吐 1TB / 900s ≈ 1138 MB/s需要同时读的节点数 ≈ 1138 / 200 ≈ 6 台 DataNode 同时打满而 Spark 的 Task 并发度取决于分区数也就是 Executor 内核数 × Executor 数量。如果每个分区处理 128MB 数据1TB 对应约 8000 个分区。要让 8000 个分区在 900 秒内跑完需要并发数 8000 / (900 / 单分区处理时间)。单分区处理时间如果是 5 秒那并发数得在 45 左右。结论就是Executor 内核总数 45 以内足够不用盲目堆资源。堆多了反而会因为 HDFS 并发读的瓶颈抬升延迟。4.2 Spark 动态资源分配处理“忙时高并行闲时省内存”的利器另一个我强烈建议开启的功能是Dynamic Resource Allocation动态资源分配。默认情况下Spark 作业启动时会一次性申请固定数量的 Executor一直占到作业结束。对于 HDFS Spark 集成场景这有个天然矛盾——白天跑批任务多Executors 全占着晚上没人跑任务集群的空闲 Executor 还在吃内存。而 HDFS 不管你有多少 Executor 在空闲它照常接受写入请求等缓存写满才落盘。启用动态分配之后Spark 会根据当前待处理任务的数量自动调整 Executor 数量忙时往上加闲时往下缩。配置如下spark.dynamicAllocation.enabledtrue spark.dynamicAllocation.initialExecutors5 spark.dynamicAllocation.minExecutors2 spark.dynamicAllocation.maxExecutors50 spark.dynamicAllocation.executorIdleTimeout60s spark.dynamicAllocation.schedulerBacklogTimeout5s这个配置特别适合“白天高峰期多个业务方轮流提交作业、晚上几乎无作业”的共享集群。我在一个实际项目里开了这个功能集群的日常资源利用率从 37% 提升到 68%跑批高峰期也没再出现资源争抢导致的延迟。但要提醒一句动态分配依赖External Shuffle ServiceESS。因为 Executor 被关闭但 Shuffle 数据还在磁盘上后续 Stage 的 Task 要去读上一个 Stage 的输出必须依赖一个常驻的 ESS 进程来提供这些数据。如果你没有做这一步直接开动态分配作业会在 Shuffle 阶段大面积报错。4.3 Shuffle 阶段与 HDFS 的关系一个需要“桥接”的环节很多人会把 Spark 的 Shuffle 和 HDFS 隔离开来看认为 Shuffle 是 Spark 内部的事。但集成环境里Shuffle 失败或变慢常常会反过来拖垮 HDFS 的性能。Shuffle Write 阶段每个 Mapper 会把输出写到本地磁盘Shuffle Read 阶段Reducer 要从所有 Mapper 所在的节点拉取这些输出文件。如果你把 Executor 内存调得很大每个 Executor 上跑好几个 Mapper然后同时还有好几个 Shuffle 输出的文件在往本地磁盘刷HDFS 的 DataNode 同时还在做容错复制写三副本本机磁盘 IO 容易瞬间冲到 100%。应对手段有几条Executor 内存尽量小于节点内存的 1/3别贪大。spark.shuffle.file.buffer可以适当增大比如从默认 32KB 调到 128KB减少写磁盘的次数直接在 Spark 层面降低磁盘 IO 压力。选择好的 Shuffle 实现。Spark 2.0 之后默认就是 Sort Shuffle Manager这个不要动。研究一下spark.shuffle.manager是不是被改回 Hash 了老版本跑批任务的一些团队会改Hash Shuffle 在 Executor 数多的时候会产生海量小文件HDFS 上没问题但本机文件句柄和磁盘寻道开销极大。三年前我用 CDH 5.x 的 Spark 1.6 时见过一个明显被坑的项目一个中等规模集群跑一个 join 类型的批处理作业全部时间都耗在 Shuffle 阶段。上去一看spark.shuffle.managerhash修改为 sort 后直接用同一份代码同一个数据量作业时间降了 60%。现在 Spark 3.x 默认已经不用配置了但如果是老集群、老项目值得检查一次。5. 小文件治理与增量数据场景当时觉得很麻烦后来发现全是收益5.1 流式写入 HDFS 导致的小文件灾难我要特别聊一个场景很多团队用 Kafka Structured Streaming 实时落数仓最终结果写到 HDFS 上。流式任务每隔几秒就写一批默认情况下每个微批会生成一批新的数据文件。跑一晚上几千甚至几万个文件就堆在 HDFS 目录里了。这些文件是 Spark 直接写出来的不是通过 Hive 事务或像 Hudi/Iceberg 这类带数据治理能力存储格式管理的。后续有人要用 Spark SQL 去读这张表整个目录扫一遍一个大目录几万个文件Spark 生成几万个分区立刻回到小文件泥潭。痛点是日积月累的等到你某天发现“批处理作业怎么越跑越慢”往往已经是两三个月之后了。你去集群上看一个分区目录下堆积了几万个小文件一个 LIST 操作就能让 NameNode 卡住半秒多张表同时来一下NameNode 直接到了“假死”的边缘。5.2 合并策略Coalesce 还是 Repartition在 Spark 中合并小文件最绕不开的无非这两个算子的问题coalesce()和repartition()。这两个都能调整输出文件数但工作方式完全不同coalesce(n)只在节点间做小规模的 shuffle比如从 2000 个分区合并到 100 个很多分区是不用动数据的直接把同一个节点的相邻分区串起来。效率高但合并后的数据可能分布不均匀。repartition(n)做一次全量 hash 重分区数据重新洗牌最终文件大小均匀但由于 Shuffle 会产生大量中间数据成本比 coalesce 高不少。我的一般选择逻辑是文件数量降到百级以内用 coalesce要求最终文件大小完全可控、均匀分布时用 repartition。写一个示例合并一张 Hive 分区表的某个分区的数据from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(compact_hdfs_partition) \ .enableHiveSupport() \ .getOrCreate() target_table ods.order_detail target_ds 20250101 # 读指定分区 df spark.sql(f SELECT * FROM {target_table} WHERE ds {target_ds} ) # 计算目标文件数分区内总数据量 / 目标单文件大小(128MB) import math total_bytes df.count() * len(df.columns) # 粗略估算实际可用 Spark plan 的行数统计 target_files max(1, math.ceil(total_bytes / (128 * 1024 * 1024))) # 如果 source 文件数已经接近 target_files跳过合并 source_files spark.sql(f SELECT COUNT(*) AS cnt FROM {target_table} WHERE ds {target_ds} ).collect()[0].cnt if source_files int(target_files * 0.8): print(无需合并) else: # 用 repartition 做重分区写入指定 parquet zstd 压缩 df.repartition(target_files) \ .write \ .mode(overwrite) \ .format(parquet) \ .option(compression, zstd) \ .insertInto(target_table, overwriteTrue)注意一个关键点覆盖写不能直接对整个分区目录 delete 然后 write。要先把新数据写到一个临时目录通过 Hive 的INSERT OVERWRITE TABLE ... PARTITION(...)的方式或者insertInto(..., overwriteTrue)来原子替换。否则覆盖写失败或者中途作业挂掉坑就很深了。5.3 增量不可变数据与“日期分区 合并”的套路在数据处理里最推荐的还是“增量写 定期合并”的模式。比如实时链路落 ODS 层按天分区。每天的数据当天生成的文件可能 5000 个跑一个定时任务在凌晨统一合并到当天分区的目标文件数量。合并完当天分区只留下 10 到 20 个 128MB 的 Parquet 文件下游 DW 层再读这层数据的时候分区的文件数和大小就可控了。这里要留意合并任务本身的资源消耗。我们用 20 个 Executor 跑一个 500GB 分区的合并任务耗时约 8-10 分钟这个成本是值得付的——下游任务原本因为小文件多要跑 1 个多小时合并后跑 20 分钟整体反而省了。如果你有更频繁的需求可以考虑引入 Hudi 或 Iceberg它们自带小文件自动合并和事务机制。但对纯 HDFS Spark 的老技术栈来说一个简单的“夜间 compaction 作业”是成本最低、最稳妥的方案。6. 一套实用的性能诊断流程肉眼排障法 面板监控法6.1 从 Spark UI 快速定位 HDFS 瓶颈我给团队做 Spark 性能调优时有一个固定的排查顺序打开 Spark UI 的 Stages 页面看哪个 Stage 耗时最长。点进该 Stage看 Input 数据量和对应的 Shuffle Read 数据量。看 Task 的 Locality Level Summary确认数据本地性分布。如果 Input 数据量异常大怀疑是分区裁剪失效或读到了多余列。如果 Task 平均执行时间短但任务数极多怀疑是小文件问题和分区数过多。看 Executor 的 GC 时间如果 GC 时间超过执行时间的 20%内存设置有问题。有一个很值得留意的指标Input Size / Records 和 Shuffle Write Size / Records的关系。如果 Shuffle 写出量是输入数据量的几倍以上那绝对是业务逻辑或 Join 设计出了问题比如笛卡尔积、分组键冲突导致数据倾斜等。6.2 命令行与监控面板HDFS 侧的诊断手段HDFS 侧的排障主要靠几个命令。排查文件分布hdfs fsck -blocks /path/to/table这条命令会输出文件数量、块数量、副本状态。如果文件数远大于块数比如文件数 5 万、块数 4000一眼就能看出小文件问题。排查节点的负载和数据本地性hdfs dfsadmin -report看每个 DataNode 的容量使用率和最后一个心跳时间。如果某些节点磁盘占用远高于其他节点说明数据倾斜严重Spark 读数据时某些节点会成为热点。追踪读慢的具体原因可以看 DataNode 日志tail -f /var/log/hadoop-hdfs/hadoop-hdfs-datanode-*.log | grep -E slow|timeout|exception如果日志里大量出现 “Slow block receiver” 或者 “write to disk timed out”说明磁盘 IO 已经到了极限。配合iostat和dstat查看系统层指标基本能判断瓶颈是CPU、内存、网络还是磁盘。6.3 一次完整排障实录从“作业慢 3 倍”到“3 条配置 1 次合并”我拿一次真实案例把整合诊断流程演示一遍。背景一个保险行业客户用它自建的 CDH 6.2 集群跑寿险保单的数据加工。某一张保单明细表的日批任务从最初 15 分钟逐步恶化到 55 分钟。客户反馈“集群加了机器没有改善”。我接手之后第一步查 Spark UI。Stage 0 的 Input 显示是 460GB——这个保单明细表在 HDFS 上的目录大小只有 89GB说明读的时候发生了重复读。点进详情一看SQL 里 join 条件设计有误把一个大表复制了一份小表 Broadcast 不成功退化成 SortMergeJoin导致中间结果放大 5 倍。第二步查文件分布。hdfs fsck /user/hive/warehouse/ods_policy_detail显示该目录下文件数 21 万个平均文件大小 4.2MB。第三步查本地性。Spark UI 显示 NODE_LOCAL 占 48%PROCESS_LOCAL 只有 2%ANY 占 29%。于是做完三件事改掉 SQL 中导致数据膨胀的 join 写法对分区目录做一次小文件合并21 万 → 2800 个文件把spark.locality.wait从默认 3s 调到 8s。结果同样的批任务耗时降到 21 分钟。后续观察了一个月稳定在 19-23 分钟。这件事给我最大的启发是性能优化不是单点作战而是链路串联。文件层面的问题、SQL 层面的问题、调度层面的问题往往同时存在。只看表面一个“慢”字不做系统排查盲调参数是没有意义的。7. 集成的进阶玩法用 Alluxio 和分层存储绕过 HDFS 的短板7.1 HDFS 在上游频繁读取场景下的“阿喀琉斯之踵”HDFS Spark 这套组合跑得很成熟之后你还是会遇到一个结构性问题HDFS 的元数据服务 NameNode 是单点既然没法在架构上引入联邦或多 NameNode 模式那在超高频读取场景里它永远是短板。比如用 Spark 做在线服务的特征计算需要每 5 分钟读一次 HDFS 上的 100GB 特征库。这种情况下直接每次从 HDFS 读NameNode 的 RPC 请求会把你打哭。解决思路不是去调 Hadoop 参数而是引入缓存层。7.2 Alluxio 作为 HDFS 与 Spark 之间的“数据高速公路”Alluxio 是什么一句话概括它是个分布式内存文件系统把数据从底层存储HDFS、S3、OSS透明缓存起来给上层计算引擎Spark、MapReduce、Flink挂载成一个 HDFS 兼容的命名空间。换句话说Spark 读的路径从 “Executor → HDFS” 变成了 “Executor → Alluxio内存/SSD 缓存 → 命中缓存就直接返回未命中才回源到 HDFS”。使用 Alluxio 的收益是什么举一个我实践过的场景某算法团队每天用 Spark 做模型样本拼接要反复读取同一份 3TB 的点击日志做特征交叉验证。原先每次读 HDFS需要 20 分钟。接入 Alluxio 之后第一次跑温数据读 HDFS 载入缓存后续每次跑热数据读全在内存里耗时压到 3 分钟以内。当然这个方案也有代价Alluxio 需要独占一组节点来跑 worker或者和 Spark Executor 混部。运维复杂度上升。如果只是偶尔跑一次批处理完全没有必要引入 Alluxio。它适合的是“同一份数据被反复读取”的场景比如特征库、维表、样本库。7.3 存储优化策略的组合使用最后我把分层存储策略再说透一点。如果已经引入 Alluxio那缓存策略就要和 HDFS 的 Storage Policy 配合经常被速查的小维表放到 Alluxio 内存缓存长驻不换出。需要全量读取做训练的数据优先放在 HDFS 的 SSD 策略下。冷数据和归档数据直接切到 COLD 策略Alluxio 不缓存读的时候接受一定延迟。我在实际项目里用这套组合策略为团队减少了对存储介质的重复采购需求——之前给 Spark 批处理预留的高性能磁盘一直不够用后来深入做冷热分层约 40% 的冷数据被迁走热数据的磁盘负载降到之前的 60%。8. 集成优化的持续运维日常巡检和避坑清单8.1 我列出的“每天必看”运维项HDFS 与 Spark 集成的优化不是一锤子买卖跑批集群的性能会随着数据量增长、业务逻辑演进而持续衰减。我建议每个团队至少把这几项作为日常巡检分区目录文件数用脚本每天扫一遍核心业务表的每个分区文件数超过阈值比如 2000就告警。Spark 作业 Input 数据量偏差同一个作业前后两天的 Input 数据量如果突然翻倍立刻追踪是哪张表的数据分布变了。NameNode RPC 延迟以 95 分位延迟为依据超过 50ms 就检查是否有小文件读取风暴或者元数据操作异常。执行器 GC 时间周期统计作业 GC/执行时间比例如果长期高于 15%要考虑调整内存或优化代码里的对象大小。数据本地性分布定期抽查核心作业的 Locality Level Summary如果 PROCESS_LOCAL 占比长期低于 30%需要排查资源调度策略。8.2 绝对不要做的事踩过的坑合集最后把我和同行反复踩过的坑集中成一条避坑清单每一条背后都有血泪教训不要在 HDFS 上存 Gzip 压缩的文本文件让 Spark 读取。不支持 split单个大文件会让 Spark 退化为单分区处理。不要重建表时选错分区字段。拿用户 ID 做分区字段的表一年下来你会得到一张几十万甚至上百万分区目录的怪物表NameNode 会持续被这份元数据拖垮。不要忽略 Seconds 级别的 GC 时间。Spark 一个长任务如果老 GC 占比高你先查代码是不是创建了海量小对象而不是急着加内存。加内存只能延缓不能根治。不要对 Executor 内存贪心。Executor 内存越大单节点上并发线程越多Shuffle 和 HDFS 写操作对磁盘的竞争越激烈。极端情况下整个集群的吞吐反而会因为“大 Executor”而下降。不要在业务高峰期跑数据合并任务。Compaction 作业会重度占用 HDFS 磁盘 IO如果同时跑大批量 Spark 批处理两边互相拖最终双双变慢。要预排任务优先级让 compaction 作业在低峰窗口执行。以上是我在 HDFS 和 Spark 集成性能优化上最核心的实战经验和方法论。每次帮客户排查我基本都会从头到尾走一遍这个链路——读数据路径、文件组织、资源调度、Shuffle 调优、小文件治理、持续巡检。这套东西覆盖了绝大多数“Spark 跑 HDFS 数据变慢”的场景。如果你手里正有一套跑批任务在变慢不妨按这个顺序一层层查大概率问题就藏在这几个常见位置里。
HDFS与Spark集成性能优化：原理、参数调优与实战诊断
1. 先搞清楚一件事HDFS 和 Spark 集成到底在解决什么问题很多人一上来就谈参数调优谈内存分配但我觉得有个更根本的问题得先聊透:HDFS 和 Spark 本来就不是很“般配”的一对它们能凑在一起工作本质上是大数据技术栈在特定历史阶段下的一种“务实选择”。我先说说这两兄弟的身世。HDFS 诞生于 Hadoop 生态它擅长的是海量文件的分布式存储——默认块大小 128MB写入一次、读取多次为了容错会把每个块复制三份。但它的计算能力几乎为零它只管存不管算。而 Spark 恰好相反它是一个内存计算框架擅长把复杂的数据处理任务拆成一个个阶段Stage去跑但它自己不带存储系统数据总得有地方放。所以在大数据平台里最常见也最经典的一种架构就是HDFS 负责存Spark 负责算两者通过网络对接。这套组合天然就有几个问题一是网络开销大。Spark 从 HDFS 读数据本质上是把远端节点的数据拉到 Executor 所在节点去处理。如果数据在节点 A而跑任务的 Executor 在节点 B那中间就隔着一次网络传输。集群规模一大节点上百台每次任务都要跨节点拉数据整体吞吐立刻就成了瓶颈。二是元数据压力。HDFS 的 NameNode 是单点它维护着整个文件系统的目录树和文件/块映射信息。Spark 提交一个作业如果读一张有几万个分区的表那它就要向 NameNode 发起海量元数据请求。三是文件大小分布不均。HDFS 不适合存大量小文件而 Spark 的 partition 数量和输入文件数量/大小直接相关文件一多且碎任务开销就成倍增长。我在实际项目中见过太多团队把 Spark 跑得慢归咎于“集群不够大”或“代码写得烂”实际上有相当一部分性能问题出在 HDFS 和 Spark 的衔接层——文件格式选错、压缩方式不对、数据没有做分区裁剪、跑任务的时候数据本地性极差。这些不是靠堆机器能解决的根子在于你没把这两个系统当做一个整体来设计。这篇文章我就围绕“HDFS Spark 集成”这条主线把我实操中积累的性能优化方法拆开揉碎了讲清楚。内容包括读数据的路径优化、文件布局与格式选择、资源调度与本地性配置、Shuffle 阶段的调优以及一套可以直接拿去用的诊断手段。每个部分我都会给到具体的配置参数和验证方法而不是只讲空洞的原理。2. 读数据链路HDFS 到 Spark 的路径上藏着最多性能损耗2.1 一次读数据的完整旅程我们先画一条线Spark 提交一个作业读取 HDFS 上一个 10GB 的文件这条链路上到底发生了什么。第一步Spark Driver 通过 Hadoop 客户端 API 向 NameNode 发起请求获取这个文件对应的 Block 位置列表。NameNode 会返回每个 Block 所在的 DataNode 列表通常是三个副本对应的节点。第二步Driver 根据这些位置信息划分 RDD 分区。默认情况下HDFS 文件被切分成的分区数 文件大小 / 块大小128MB。每个分区对应一个或多个 Block 的连续区间。第三步Executor 启动 TaskTask 内部通过 Hadoop InputFormat 去 DataNode 拉取数据。这里的关键动作叫“数据本地性调度”如果 Task 被分配到了一个 Block 副本所在的节点上那它走的是本地读路径Short-Circuit Local Reads直接读本地磁盘文件速度快到惊人如果 Task 被分配到其他节点那就要走 DataNode 的 RPC 接口把 Block 数据通过网络传过去。第四步数据进入 Spark 的 Memory 或 DiskStore然后被反序列化成 Java 对象进入 RDD 的计算管线。链路不长但每一步都有可优化的空间。我按我踩坑的严重程度排个序。2.2 数据本地性90% 的人没认真查过的配置先说影响最直接的一个数据本地性Data Locality。Spark 的 TaskScheduler 在分发 Task 时会根据 DAGScheduler 提交的 taskset 里标注的 preferredLocations 去匹配 Executor。它会按PROCESS_LOCAL进程内本地→ NODE_LOCAL同节点本地→ RACK_LOCAL同机架→ ANY跨网络的顺序依次尝试。默认的spark.locality.wait是 3 秒也就是说如果数据本地性达不到较优的级别调度器会在该级别等待 3 秒然后再放宽。生产环境里常见的现象是Task 迟迟起不来日志里大量出现 “Locality Level: ANY”跑完之后看监控面板输入数据全是通过远程网络传输的。我遇到过最夸张的一次一个 1TB 的 Hive 表做聚合集群 50 个节点跑了一个小时还没结束。后来查 Spark UI发现几乎所有 Task 的本地性都是 NODE_LOCAL 和 RACK_LOCALPROCESS_LOCAL 的比例不到 5%。原因是什么执行器的内存里缓存了大量旧任务的 RDD新任务的输入数据虽然在同节点但 Executor 没资源了Task 被放到了其他节点。排查方法很简单在 Spark UI 的 Stages 页面点开任意一个 Stage看 “Locality Level Summary” 这个表格。如果发现全是 ANY 或者 RACK_LOCAL 占大头那就是本地性出了问题。常规解法有这么几种增大 Executor 内存避免因为内存不足把缓存刷掉导致重复计算。调整spark.locality.wait比如从 3 改成 5 甚至 10给调度器更多时间去匹配本地节点。别小看这个参数我曾经把一批小时级作业的耗时从 50 分钟压到 28 分钟就靠这一条。检查 input split 的划分方式。HDFS 的块和 Spark 的输入分区不总是一一对应的spark.hadoop.mapreduce.input.fileinputformat.split.maxsize如果被改了会导致跨节点读数据破坏本地性。还有一个容易忽略的点开启 HDFS Short-Circuit Local Reads。这个功能让 DataNode 上的数据读取绕过 TCP 栈直接用本地文件描述符来读。需要配置 DataNode 的dfs.domain.socket.path同时把dfs.client.read.shortcircuit打开。在 CDH 和 HDP 发行版里这是默认开启的但自己用手工方式搭的 Hadoop 集群比如 Apache 纯原版经常是关着的。要验证当前是否开启可以在 DataNode 日志里搜 “shortcircuit”或者跑一个本地读的任务看数据吞吐量是否异常低。2.3 文件格式和压缩同一个文件差 3 到 8 倍的读取速度这是集成层面能拿到最大收益、但改动成本最低的一部分。HDFS 上存的文件格式直接决定了 Spark 读取时要做多少额外工作。常见的有文本文件TextFile、SequenceFile、Parquet、ORC、Avro 这么几种。我的结论非常明确只要不是没办法生产环境一律用 Parquet 或 ORC别用文本文件。为什么因为列式存储有两个属性对 Spark 极有价值。第一个是谓词下推Predicate Pushdown。也就是说查一张有 100 个字段的表过滤条件是where a5如果文件是列式存储Spark 只需要读取 a 字段和行号索引那一列的数据其他 99 列物理上根本不用加载。这对 IO 的节省不是一倍两倍是数量级的。第二个是压缩比。Parquet 配合 Snappy 或 ZSTD 压缩对结构化数据通常能做到存储压缩比在 5:1 以上。HDFS 上存的文件小读的时候网络传输也少整个链路都快。我放个实测数据表格是之前做的一个电商订单分析任务同样一份数据量约 120GB 的订单明细存储格式压缩方式HDFS 占用全量读取耗时Spark 2.4.8 / 20 个 ExecutorSQL 过滤后再聚合耗时文本 CSV无120GB47 分钟39 分钟文本 CSVGzip35GB22 分钟18 分钟ParquetSnappy22GB9 分钟4 分钟ParquetZSTD18GB8 分钟3 分钟注意 Gzip 文本那一栏——虽然文件变小了读取却并不快。因为 Gzip 压缩的文本文件不支持 splitSpark 只能单线程去解压整个文件等于把一个几 GB 的文件变成了一个无法切分的巨型分区。所以文本文件用 Gzip 压缩在 Spark 场景下是反模式我见过不少团队因为这个原因作业跑得特别慢。Parquet 本身是 splittable 的配合压缩不会破坏并发度。ZSTD 压缩比和速度都比 Snappy 好一点如果你的集群是 CDH 6.x 之后或者 Apache Spark 3.x我建议直接上 ZSTD。2.4 小文件问题HDFS 块大小和 Spark 分区数之间的博弈这个问题我单独拿出来说因为它是 HDFS 和 Spark 集成的“经典老坑”。HDFS 默认块大小是 128MBNameNode 存储元数据的压力决定了它不适合存海量小文件。而 Spark 读取 HDFS 上的一个目录时是一个一个文件去切分的每个小文件都会至少生成一个分区。假设你在 HDFS 上存了 10 万个小文件每个文件 100KB总量约 10GB。Spark 读这个目录会生成 10 万个小分区每个分区就处理 100KB 数据。这是个灾难——10 万个 Task 光调度开销就比数据处理本身还高而且会产生 10 万次对 NameNode 的元数据请求NameNode 通常会因为 RPC 队列爆掉而开始卡顿。解决方案分三个层面写入时治理写数据的时候控制文件数量。用 Spark 写 HDFS 时通过coalesce()或repartition()控制输出文件数目标让每个输出文件接近 128MB。定期合并用 Spark 作业对历史小文件目录做一次“Compact”操作按分区读取后重新写回。写回时利用分区键让数据均匀分布到合理的文件数量。动态分区裁剪读取时用 Hive 分区表的方式组织数据让 Spark 通过分区剪枝只扫描需要的分片目录。我之前给一个项目做过一次存储盘点和整合把 60 万个小文件合并到约 8000 个 128MB 以上的 Parquet 文件。做完之后同一个下游 Spark 批处理作业从 2 小时 10 分钟降到 26 分钟NameNode 的心跳稳定性也上来了。这个优化在整个链路里性价比极高。3. 建表与数据组织给 Spark 一个“对胃口”的 HDFS 目录结构3.1 从 Hive 表到 HDFS 目录一张表就是一个业务契约在很多团队里Spark 读 HDFS 数据不是直接指定路径的而是通过 Hive Metastore 里的表对象来读取。这里有个概念需要点透Hive 表在 HDFS 上的存储目录结构基本决定了 Spark SQL 的性能上限。一张 Hive 表在 HDFS 上的物理形态就是一个目录里面放着数据文件。分区表会在这个目录下再按分区字段建子目录比如order_ds20250101/、order_ds20250102/。这一步非常关键——Spark 在跑 SQL 的时候利用分区信息做裁剪直接决定扫描多少数据。我最常见到的错误用法是建表的时候没有做分区设计或者分区字段选错了。分区字段要选“低基数的过滤字段”比如日期、地区、业务线。不要选用户 ID、订单号这种高基数字段那会导致每个分区数据量极小而分区数量爆炸。分区粒度也要想清楚。你按天分区还是按小时分区如果按小时分区一年的数据就是 8760 个分区目录。对于一张日增 200GB 的表按小时分区还勉强能接受但如果日增只有 20GB一个小时分区只有不到 1GB 数据那分区裁剪的意义就变小了反而增加了元数据开销。我的经验值是单个分区内的数据量至少要在 5GB 以上最好在 50GB 到 500GB 之间。3.2 分区、分桶与排序让数据分布匹配查询模式分桶Bucket是比分区更细粒度的文件组织方式。在 Spark 3.0 之后分桶表配合bucket pruning能实现类似“预聚合”的效果——你不知道具体在哪个分区但通过分桶字段的哈希值能直接定位到对应的桶文件去读。这在相关性 Join比如大表关联小表场景下能显著减少 Shuffle 数据量。但说实话我给你一句真实的建议如果你不是有明确的高频 Join 场景不要轻易引入分桶。分桶表写入逻辑复杂对并发写不友好配合不好反而得不偿失。更实用的技巧是布局排序在写 Parquet 文件之前先按常用的过滤字段比如订单日期、商品类目做一次全局排序。排序之后的数据在写入 Parquet 时min/max 统计信息才能充分发挥作用——Parquet 每个 RowGroup 都会记录该列的最小值和最大值查询时如果过滤条件命中不到 RowGroup 的范围整个 RowGroup 直接跳过。不排序的情况下每个 RowGroup 的 min/max 范围都很大统计信息基本等于废的。我曾经做过一个线上 A/B 对比同一张 300GB 订单表数据文件一个做了全局排序再写 Parquet一个直接写。跑同样的“按用户 ID 查最近 100 条订单”查询排序后的版本扫描的数据量只有未排序版本的 1/5。因为用户 ID 在某一个时间范围内往往集中在某些文件里排序让这些数据物理上靠在一起了。所以我把这个环节总结成一个闭环先用 Hive 分区把时间维度卡死再用 Parquet 列式存储降低列扫描 IO最后在写入前按常用过滤字段排序激活 min/max 裁剪。这三板斧下来绝大多数查询性能都能有个量级上的提升。3.3 存储策略冷热分离和分层存储到了集群运维层面还有个容易被忽视的点HDFS 上的数据不是永远“热”的。Spark 跑批处理既要读最近一天的数据也可能要读三个月前、一年前的数据。如果你把所有数据都放在内存计算集群默认读取的存储策略下集群磁盘 IO 和网络带宽会被“冷数据查询”持续拖累。这里有一个成熟的做法利用 HDFS 的Storage Policy存储策略做冷热分层。比如把近 7 天的数据放到ALL_SSD策略下把 7 天到 90 天的数据放到HOT默认三副本 HDD策略下超过 90 天的数据放到COLD比如归档节点或冷备存储策略下。Spark 读数据时因为 DataNode 会根据策略调度数据到不同介质上热的查询走 SSD冷的查询走 HDD互不干扰。我建议运维同学在生产环境把这套策略落实下去。虽然这是偏向平台侧的优化影响却是实打实的——跑批任务高峰期的平均 IO 延迟能下降 20% 到 40%。4. 资源调度与动态分配让 Executor 和 DataNode 真正“对齐”4.1 Executor 数量、内存分配和并行度的三角平衡聊 HDFS 与 Spark 集成资源调度是逃不开的一环。因为 Spark 的 Executor 数量和每个 Executor 上跑的 Task 并行度直接决定了它到底能同时向 HDFS 发起多少个并发读取流。并行度不够哪怕 HDFS 底层盘阵性能再好Spark 也只能把吞吐打一半并行度太高DataNode 的磁盘 IO 和网络带宽先被打爆NameNode 的 RPC 请求会堆积。我的实践策略是先算并发数再定资源量。假设一个批处理任务要读 1TB 的 Parquet 数据HDFS 平均单节点磁盘吞吐按 200MB/s 估算机械盘阵列SSD 会更好如果希望 15 分钟内完成读取需要的吞吐 1TB / 900s ≈ 1138 MB/s需要同时读的节点数 ≈ 1138 / 200 ≈ 6 台 DataNode 同时打满而 Spark 的 Task 并发度取决于分区数也就是 Executor 内核数 × Executor 数量。如果每个分区处理 128MB 数据1TB 对应约 8000 个分区。要让 8000 个分区在 900 秒内跑完需要并发数 8000 / (900 / 单分区处理时间)。单分区处理时间如果是 5 秒那并发数得在 45 左右。结论就是Executor 内核总数 45 以内足够不用盲目堆资源。堆多了反而会因为 HDFS 并发读的瓶颈抬升延迟。4.2 Spark 动态资源分配处理“忙时高并行闲时省内存”的利器另一个我强烈建议开启的功能是Dynamic Resource Allocation动态资源分配。默认情况下Spark 作业启动时会一次性申请固定数量的 Executor一直占到作业结束。对于 HDFS Spark 集成场景这有个天然矛盾——白天跑批任务多Executors 全占着晚上没人跑任务集群的空闲 Executor 还在吃内存。而 HDFS 不管你有多少 Executor 在空闲它照常接受写入请求等缓存写满才落盘。启用动态分配之后Spark 会根据当前待处理任务的数量自动调整 Executor 数量忙时往上加闲时往下缩。配置如下spark.dynamicAllocation.enabledtrue spark.dynamicAllocation.initialExecutors5 spark.dynamicAllocation.minExecutors2 spark.dynamicAllocation.maxExecutors50 spark.dynamicAllocation.executorIdleTimeout60s spark.dynamicAllocation.schedulerBacklogTimeout5s这个配置特别适合“白天高峰期多个业务方轮流提交作业、晚上几乎无作业”的共享集群。我在一个实际项目里开了这个功能集群的日常资源利用率从 37% 提升到 68%跑批高峰期也没再出现资源争抢导致的延迟。但要提醒一句动态分配依赖External Shuffle ServiceESS。因为 Executor 被关闭但 Shuffle 数据还在磁盘上后续 Stage 的 Task 要去读上一个 Stage 的输出必须依赖一个常驻的 ESS 进程来提供这些数据。如果你没有做这一步直接开动态分配作业会在 Shuffle 阶段大面积报错。4.3 Shuffle 阶段与 HDFS 的关系一个需要“桥接”的环节很多人会把 Spark 的 Shuffle 和 HDFS 隔离开来看认为 Shuffle 是 Spark 内部的事。但集成环境里Shuffle 失败或变慢常常会反过来拖垮 HDFS 的性能。Shuffle Write 阶段每个 Mapper 会把输出写到本地磁盘Shuffle Read 阶段Reducer 要从所有 Mapper 所在的节点拉取这些输出文件。如果你把 Executor 内存调得很大每个 Executor 上跑好几个 Mapper然后同时还有好几个 Shuffle 输出的文件在往本地磁盘刷HDFS 的 DataNode 同时还在做容错复制写三副本本机磁盘 IO 容易瞬间冲到 100%。应对手段有几条Executor 内存尽量小于节点内存的 1/3别贪大。spark.shuffle.file.buffer可以适当增大比如从默认 32KB 调到 128KB减少写磁盘的次数直接在 Spark 层面降低磁盘 IO 压力。选择好的 Shuffle 实现。Spark 2.0 之后默认就是 Sort Shuffle Manager这个不要动。研究一下spark.shuffle.manager是不是被改回 Hash 了老版本跑批任务的一些团队会改Hash Shuffle 在 Executor 数多的时候会产生海量小文件HDFS 上没问题但本机文件句柄和磁盘寻道开销极大。三年前我用 CDH 5.x 的 Spark 1.6 时见过一个明显被坑的项目一个中等规模集群跑一个 join 类型的批处理作业全部时间都耗在 Shuffle 阶段。上去一看spark.shuffle.managerhash修改为 sort 后直接用同一份代码同一个数据量作业时间降了 60%。现在 Spark 3.x 默认已经不用配置了但如果是老集群、老项目值得检查一次。5. 小文件治理与增量数据场景当时觉得很麻烦后来发现全是收益5.1 流式写入 HDFS 导致的小文件灾难我要特别聊一个场景很多团队用 Kafka Structured Streaming 实时落数仓最终结果写到 HDFS 上。流式任务每隔几秒就写一批默认情况下每个微批会生成一批新的数据文件。跑一晚上几千甚至几万个文件就堆在 HDFS 目录里了。这些文件是 Spark 直接写出来的不是通过 Hive 事务或像 Hudi/Iceberg 这类带数据治理能力存储格式管理的。后续有人要用 Spark SQL 去读这张表整个目录扫一遍一个大目录几万个文件Spark 生成几万个分区立刻回到小文件泥潭。痛点是日积月累的等到你某天发现“批处理作业怎么越跑越慢”往往已经是两三个月之后了。你去集群上看一个分区目录下堆积了几万个小文件一个 LIST 操作就能让 NameNode 卡住半秒多张表同时来一下NameNode 直接到了“假死”的边缘。5.2 合并策略Coalesce 还是 Repartition在 Spark 中合并小文件最绕不开的无非这两个算子的问题coalesce()和repartition()。这两个都能调整输出文件数但工作方式完全不同coalesce(n)只在节点间做小规模的 shuffle比如从 2000 个分区合并到 100 个很多分区是不用动数据的直接把同一个节点的相邻分区串起来。效率高但合并后的数据可能分布不均匀。repartition(n)做一次全量 hash 重分区数据重新洗牌最终文件大小均匀但由于 Shuffle 会产生大量中间数据成本比 coalesce 高不少。我的一般选择逻辑是文件数量降到百级以内用 coalesce要求最终文件大小完全可控、均匀分布时用 repartition。写一个示例合并一张 Hive 分区表的某个分区的数据from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(compact_hdfs_partition) \ .enableHiveSupport() \ .getOrCreate() target_table ods.order_detail target_ds 20250101 # 读指定分区 df spark.sql(f SELECT * FROM {target_table} WHERE ds {target_ds} ) # 计算目标文件数分区内总数据量 / 目标单文件大小(128MB) import math total_bytes df.count() * len(df.columns) # 粗略估算实际可用 Spark plan 的行数统计 target_files max(1, math.ceil(total_bytes / (128 * 1024 * 1024))) # 如果 source 文件数已经接近 target_files跳过合并 source_files spark.sql(f SELECT COUNT(*) AS cnt FROM {target_table} WHERE ds {target_ds} ).collect()[0].cnt if source_files int(target_files * 0.8): print(无需合并) else: # 用 repartition 做重分区写入指定 parquet zstd 压缩 df.repartition(target_files) \ .write \ .mode(overwrite) \ .format(parquet) \ .option(compression, zstd) \ .insertInto(target_table, overwriteTrue)注意一个关键点覆盖写不能直接对整个分区目录 delete 然后 write。要先把新数据写到一个临时目录通过 Hive 的INSERT OVERWRITE TABLE ... PARTITION(...)的方式或者insertInto(..., overwriteTrue)来原子替换。否则覆盖写失败或者中途作业挂掉坑就很深了。5.3 增量不可变数据与“日期分区 合并”的套路在数据处理里最推荐的还是“增量写 定期合并”的模式。比如实时链路落 ODS 层按天分区。每天的数据当天生成的文件可能 5000 个跑一个定时任务在凌晨统一合并到当天分区的目标文件数量。合并完当天分区只留下 10 到 20 个 128MB 的 Parquet 文件下游 DW 层再读这层数据的时候分区的文件数和大小就可控了。这里要留意合并任务本身的资源消耗。我们用 20 个 Executor 跑一个 500GB 分区的合并任务耗时约 8-10 分钟这个成本是值得付的——下游任务原本因为小文件多要跑 1 个多小时合并后跑 20 分钟整体反而省了。如果你有更频繁的需求可以考虑引入 Hudi 或 Iceberg它们自带小文件自动合并和事务机制。但对纯 HDFS Spark 的老技术栈来说一个简单的“夜间 compaction 作业”是成本最低、最稳妥的方案。6. 一套实用的性能诊断流程肉眼排障法 面板监控法6.1 从 Spark UI 快速定位 HDFS 瓶颈我给团队做 Spark 性能调优时有一个固定的排查顺序打开 Spark UI 的 Stages 页面看哪个 Stage 耗时最长。点进该 Stage看 Input 数据量和对应的 Shuffle Read 数据量。看 Task 的 Locality Level Summary确认数据本地性分布。如果 Input 数据量异常大怀疑是分区裁剪失效或读到了多余列。如果 Task 平均执行时间短但任务数极多怀疑是小文件问题和分区数过多。看 Executor 的 GC 时间如果 GC 时间超过执行时间的 20%内存设置有问题。有一个很值得留意的指标Input Size / Records 和 Shuffle Write Size / Records的关系。如果 Shuffle 写出量是输入数据量的几倍以上那绝对是业务逻辑或 Join 设计出了问题比如笛卡尔积、分组键冲突导致数据倾斜等。6.2 命令行与监控面板HDFS 侧的诊断手段HDFS 侧的排障主要靠几个命令。排查文件分布hdfs fsck -blocks /path/to/table这条命令会输出文件数量、块数量、副本状态。如果文件数远大于块数比如文件数 5 万、块数 4000一眼就能看出小文件问题。排查节点的负载和数据本地性hdfs dfsadmin -report看每个 DataNode 的容量使用率和最后一个心跳时间。如果某些节点磁盘占用远高于其他节点说明数据倾斜严重Spark 读数据时某些节点会成为热点。追踪读慢的具体原因可以看 DataNode 日志tail -f /var/log/hadoop-hdfs/hadoop-hdfs-datanode-*.log | grep -E slow|timeout|exception如果日志里大量出现 “Slow block receiver” 或者 “write to disk timed out”说明磁盘 IO 已经到了极限。配合iostat和dstat查看系统层指标基本能判断瓶颈是CPU、内存、网络还是磁盘。6.3 一次完整排障实录从“作业慢 3 倍”到“3 条配置 1 次合并”我拿一次真实案例把整合诊断流程演示一遍。背景一个保险行业客户用它自建的 CDH 6.2 集群跑寿险保单的数据加工。某一张保单明细表的日批任务从最初 15 分钟逐步恶化到 55 分钟。客户反馈“集群加了机器没有改善”。我接手之后第一步查 Spark UI。Stage 0 的 Input 显示是 460GB——这个保单明细表在 HDFS 上的目录大小只有 89GB说明读的时候发生了重复读。点进详情一看SQL 里 join 条件设计有误把一个大表复制了一份小表 Broadcast 不成功退化成 SortMergeJoin导致中间结果放大 5 倍。第二步查文件分布。hdfs fsck /user/hive/warehouse/ods_policy_detail显示该目录下文件数 21 万个平均文件大小 4.2MB。第三步查本地性。Spark UI 显示 NODE_LOCAL 占 48%PROCESS_LOCAL 只有 2%ANY 占 29%。于是做完三件事改掉 SQL 中导致数据膨胀的 join 写法对分区目录做一次小文件合并21 万 → 2800 个文件把spark.locality.wait从默认 3s 调到 8s。结果同样的批任务耗时降到 21 分钟。后续观察了一个月稳定在 19-23 分钟。这件事给我最大的启发是性能优化不是单点作战而是链路串联。文件层面的问题、SQL 层面的问题、调度层面的问题往往同时存在。只看表面一个“慢”字不做系统排查盲调参数是没有意义的。7. 集成的进阶玩法用 Alluxio 和分层存储绕过 HDFS 的短板7.1 HDFS 在上游频繁读取场景下的“阿喀琉斯之踵”HDFS Spark 这套组合跑得很成熟之后你还是会遇到一个结构性问题HDFS 的元数据服务 NameNode 是单点既然没法在架构上引入联邦或多 NameNode 模式那在超高频读取场景里它永远是短板。比如用 Spark 做在线服务的特征计算需要每 5 分钟读一次 HDFS 上的 100GB 特征库。这种情况下直接每次从 HDFS 读NameNode 的 RPC 请求会把你打哭。解决思路不是去调 Hadoop 参数而是引入缓存层。7.2 Alluxio 作为 HDFS 与 Spark 之间的“数据高速公路”Alluxio 是什么一句话概括它是个分布式内存文件系统把数据从底层存储HDFS、S3、OSS透明缓存起来给上层计算引擎Spark、MapReduce、Flink挂载成一个 HDFS 兼容的命名空间。换句话说Spark 读的路径从 “Executor → HDFS” 变成了 “Executor → Alluxio内存/SSD 缓存 → 命中缓存就直接返回未命中才回源到 HDFS”。使用 Alluxio 的收益是什么举一个我实践过的场景某算法团队每天用 Spark 做模型样本拼接要反复读取同一份 3TB 的点击日志做特征交叉验证。原先每次读 HDFS需要 20 分钟。接入 Alluxio 之后第一次跑温数据读 HDFS 载入缓存后续每次跑热数据读全在内存里耗时压到 3 分钟以内。当然这个方案也有代价Alluxio 需要独占一组节点来跑 worker或者和 Spark Executor 混部。运维复杂度上升。如果只是偶尔跑一次批处理完全没有必要引入 Alluxio。它适合的是“同一份数据被反复读取”的场景比如特征库、维表、样本库。7.3 存储优化策略的组合使用最后我把分层存储策略再说透一点。如果已经引入 Alluxio那缓存策略就要和 HDFS 的 Storage Policy 配合经常被速查的小维表放到 Alluxio 内存缓存长驻不换出。需要全量读取做训练的数据优先放在 HDFS 的 SSD 策略下。冷数据和归档数据直接切到 COLD 策略Alluxio 不缓存读的时候接受一定延迟。我在实际项目里用这套组合策略为团队减少了对存储介质的重复采购需求——之前给 Spark 批处理预留的高性能磁盘一直不够用后来深入做冷热分层约 40% 的冷数据被迁走热数据的磁盘负载降到之前的 60%。8. 集成优化的持续运维日常巡检和避坑清单8.1 我列出的“每天必看”运维项HDFS 与 Spark 集成的优化不是一锤子买卖跑批集群的性能会随着数据量增长、业务逻辑演进而持续衰减。我建议每个团队至少把这几项作为日常巡检分区目录文件数用脚本每天扫一遍核心业务表的每个分区文件数超过阈值比如 2000就告警。Spark 作业 Input 数据量偏差同一个作业前后两天的 Input 数据量如果突然翻倍立刻追踪是哪张表的数据分布变了。NameNode RPC 延迟以 95 分位延迟为依据超过 50ms 就检查是否有小文件读取风暴或者元数据操作异常。执行器 GC 时间周期统计作业 GC/执行时间比例如果长期高于 15%要考虑调整内存或优化代码里的对象大小。数据本地性分布定期抽查核心作业的 Locality Level Summary如果 PROCESS_LOCAL 占比长期低于 30%需要排查资源调度策略。8.2 绝对不要做的事踩过的坑合集最后把我和同行反复踩过的坑集中成一条避坑清单每一条背后都有血泪教训不要在 HDFS 上存 Gzip 压缩的文本文件让 Spark 读取。不支持 split单个大文件会让 Spark 退化为单分区处理。不要重建表时选错分区字段。拿用户 ID 做分区字段的表一年下来你会得到一张几十万甚至上百万分区目录的怪物表NameNode 会持续被这份元数据拖垮。不要忽略 Seconds 级别的 GC 时间。Spark 一个长任务如果老 GC 占比高你先查代码是不是创建了海量小对象而不是急着加内存。加内存只能延缓不能根治。不要对 Executor 内存贪心。Executor 内存越大单节点上并发线程越多Shuffle 和 HDFS 写操作对磁盘的竞争越激烈。极端情况下整个集群的吞吐反而会因为“大 Executor”而下降。不要在业务高峰期跑数据合并任务。Compaction 作业会重度占用 HDFS 磁盘 IO如果同时跑大批量 Spark 批处理两边互相拖最终双双变慢。要预排任务优先级让 compaction 作业在低峰窗口执行。以上是我在 HDFS 和 Spark 集成性能优化上最核心的实战经验和方法论。每次帮客户排查我基本都会从头到尾走一遍这个链路——读数据路径、文件组织、资源调度、Shuffle 调优、小文件治理、持续巡检。这套东西覆盖了绝大多数“Spark 跑 HDFS 数据变慢”的场景。如果你手里正有一套跑批任务在变慢不妨按这个顺序一层层查大概率问题就藏在这几个常见位置里。
