★ FEATURED ARTICLE HDFS里躺着一堆zip下游却在等数据这是搞数据管道最常遇到的窘境。以前我处理这类需求都是先把zip从集群里拖到本地机器解压完再传回去一次两次还能忍但文件多了之后时间和磁盘都受不了。后来我琢磨着直接在Hadoop上把zip解压到HDFS里省掉中间那趟折腾效果确实不错。这篇文章会把我的思路、选型过程、核心代码和踩过的坑完整分享出来给同样被压缩包归档逻辑困扰的朋友做个参考。无论你是数据工程师、平台维护人员还是刚接触HDFS生态的开发者只要能看懂基础Java或Shell下面这套方案就能直接改改拿去用。1. 直接解压的整体设计与方案选型1.1 为什么不能简单“下载到本地解压再回传”很多人第一反应是zip压缩包在HDFS上解压不是很简单吗hadoop fs -get到本地unzip解压再hadoop fs -put回去就完事了。小文件确实可以这么干但一旦数据量上来问题就非常明显。第一带宽和本地磁盘开销。一份zip可能几十GB下载到本地解压之后解压出来可能是几百GB甚至上TB本地磁盘一下子就被打满了。就算机器性能再好这个流程也是既要网络带宽、又要本地存储、还要CPU解压三样资源同时吃紧。第二运维操作繁琐。如果每天都有新的zip包进入HDFS你就得反复手动或写脚本重复“下载—解压—上传”每一步都可能出错传到一半断网、磁盘写满、权限不对哪个问题都够人喝一壶。第三与后续计算链路割裂。数据从HDFS取出再送回HDFS中间经过了外部环境整个链路的安全性和一致性都变差了。团队里如果有人误操作改了本地文件回传上去的内容和原包是否一致根本没人敢保证。所以最合理的思路是解压操作直接在HDFS上完成压缩包从哪个目录进解压产物以目标文件格式写到另一个HDFS目录整个过程不离开Hadoop集群。这样既能保住数据本地性又能把解压任务放到离数据更近的计算节点上执行。1.2 直接解压要解决的三个底层问题直接解压方案看着简单真上手会发现自己得先想清楚三个底层问题。第一个是数据本地性调度问题。HDFS天然会把大文件按块分布到不同节点但zip压缩包内部是顺序结构的Hadoop默认的TextInputFormat不能把它当成可切分的纯文本读。如果想让MapReduce或Spark来处理必须明确告诉框架“每个zip包是一个整体只能分给一个处理单元”同时尽量把任务调度到持有该zip块数据的节点上。这就要涉及自定义InputFormat或妥善使用二进制读取接口。第二个是zip格式的随机访问限制。zip不是简单的字节流压缩它内部由多个entry组成每个entry都有自己的局部文件头、压缩数据和中央目录。想解压某个entry往往需要从压缩包的中央目录定位偏移量或者借助ZipInputStream顺序读取。这意味着我们不能像解单个gzip文件那样把整个压缩包一次性映射成一个连续流而是要在读取过程中维护好entry级别的状态。第三个是元数据和目录结构要不要保留。zip里经常存在多级目录嵌套解压时要保留原有目录树还是直接把所有文件平铺到目标目录这是一个需要提前和业务方对齐的设计决策。如果下游任务只是扫描某一类文件平铺会方便很多如果业务方要求原始包内的目录结构就要在代码里解析每个entry的路径并逐级创建目录。这一点没确认好解压完发现文件位置全不对返工成本很高。1.3 四套可行方案的特点对比针对在HDFS上直接在跑批场景直接解压zip我实践下来有四套方案比较靠谱MapReduce任务解压、Spark分布式解压、HDFS Shell配合Streaming解压、Oozie定时调度解压。先把对比列出来再聊怎么选。方案并发能力实现复杂度适用场景典型短板MapReduce自定义解压高中大批量zip归档解压需要写Java代码Spark解压后即算高中解压完立刻做聚合分析小文件过多时性能损耗HDFS Shell unzip低低临时处理几个小包需要本地磁盘中转Oozie定时任务中中每天固定时段解压不适合交互式快速操作我用得最多的是第一种MapReduce方案原因是它最贴近HDFS原生生态不需要额外维护Spark作业队列调度稳定资源控制也直接在YARN配置里搞定。如果业务背景是“解压完马上加载进Spark做计算”那第二种更合适毕竟少一次写读。第三种的定位更接近于应急处理完全不建议放在生产环境。第四种是把“解压”编排进定时流水线时才会用依赖你在Oozie里配好工作流。2. 核心实操基于MapReduce的HDFS ZIP解压2.1 任务整体设计与必要依赖等到我把方案定为“写一个MapReducer来解压zip”第一步就是整理任务边界。我的做法是输入路径只指向存zip包的目录每个zip包作为一个map的输入splitmap里负责把zip里的每个entry解出来写到指定输出目录输出目录按照业务日期或者包名动态生成。这种设计下map数量等于zip包数量每个包之间天然隔离单个包解压失败不会影响其他包。context.progress()放在entry级别可以防止长时解压被YARN误判为任务超时这个细节太重要了后面会单独展开。依赖方面普通Hadoop环境自带的hadoop-client库足够用核心就是要引这么一串dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.6/version /dependencyHadoop 3.x的依赖包已经自带commons-compress等实用库Java自带的java.util.zip.ZipInputStream也在标准库中不需要额外引第三方解压框架。2.2 自定义InputFormat的核心实现思路要让每个zip包只被一个map处理最简洁的办法就是自定义一个InputFormat。我在项目里直接继承FileInputFormat重写它的isSplitable方法返回false告诉Hadoop“我这个输入路径上的文件不能切分”。这样框架就会为每个zip文件生成一个FileSplit一个split对应一个map。然后需要提供一个自定义RecordReader这个Reader的逻辑不复杂初次调用nextKeyValue时让key等于当前zip包的绝对路径字符串value用NullWritable占位后续调用直接返回false表示这个map只处理一个包。示意代码如下public class ZipFileRecordReader extends RecordReaderText, NullWritable { private Text key; private boolean processed false; Override public void initialize(InputSplit split, TaskAttemptContext context) { FileSplit fileSplit (FileSplit) split; key new Text(fileSplit.getPath().toString()); } Override public boolean nextKeyValue() { if (!processed) { processed true; return true; } return false; } Override public Text getCurrentKey() { return key; } Override public NullWritable getCurrentValue() { return NullWritable.get(); } Override public float getProgress() { return processed ? 1.0f : 0.0f; } Override public void close() {} }这个方案的妙处在于你不用自己做复杂的split解析Hadoop的FileInputFormat已经把文件路径和块信息都包装好了我们只是把“一个文件一条记录”的逻辑做进Reader而已。2.3 Mapper内部解压一次打开、逐Entry写HDFS重头戏在Mapper里。每个Mapper只收到一个zip包路径接下来就是打开HDFS上的文件流用ZipInputStream逐条读取entry并把每个entry写入目标输出文件。我写了一个通用Mapper核心逻辑如下public static class ZipDecompressMapper extends MapperText, NullWritable, NullWritable, NullWritable { Override protected void map(Text key, NullWritable value, Context context) throws IOException, InterruptedException { Path zipPath new Path(key.toString()); String targetDir context.getConfiguration().get(zip.target.dir); FileSystem fs zipPath.getFileSystem(context.getConfiguration()); try (FSDataInputStream in fs.open(zipPath); ZipInputStream zis new ZipInputStream(new BufferedInputStream(in))) { ZipEntry entry; while ((entry zis.getNextEntry()) ! null) { if (entry.isDirectory()) { continue; } Path outPath buildOutputPath(zipPath, entry.getName(), targetDir, fs); // 每处理完一个entry就向框架报告一次进度 context.progress(); context.setStatus(decompressing: entry.getName()); try (FSDataOutputStream out fs.create(outPath, true)) { IOUtils.copyBytes(zis, out, 4 * 1024, false); } } } } }这里有几个值得单独说明的细节。第一fs.open返回的是FSDataInputStream它支持Seek但ZipInputStream需要顺序读取所以外层包了BufferedInputStream避免每个字节都产生一次网络请求。第二entry.isDirectory()不能漏否则你会试图给目录也创建空文件白白增加大量无意义的HDFS文件数。第三我用fs.create(outPath, true)覆盖写入这样即使重试或多次执行任务也不会因为输出文件已存在而失败。第四IOUtils.copyBytes是Hadoop自带的工具方法第二个参数false表示不关闭输入流这个很关键因为关闭了zis整个循环就没法继续了。2.4 目标路径生成策略与同名文件冲突处理解压输出路径怎么定直接决定下游任务好不好用。我总结了一套规则推荐给团队后一直沿用。简单情况下把zip包名去掉.zip后缀作为目录名比如/data/raw/2024/06/order_20240601.zip解压输出就是/data/unzip/order_20240601/zip内部的目录结构原样保留。这样每个包对应一个目录路径清晰排查问题也容易。如果业务上希望把所有包的内容平铺进同一个目录那就要小心同名文件冲突。两个zip包可能都有data.csv直接都写到同一路径就会互相覆盖。我的处理方法是在输出文件名前拼接包名例如/data/unzip/20240601/order_20240601_data.csv。如果业务方有特殊要求也可以对重复文件加时间戳后缀。但不要等到冲突发生了再去想方案写代码前先确认下游是“按文件名读取”还是“按目录读取”这直接决定路径设计。一个经验是为保险起见压缩包内部如果有明显的主版本号或日期解压目录最好是带上下文的否则后续做增量同步时根本没法区分更新批次。3. 进阶玩法把ZIP当输入源直接参与计算3.1 不是“先解压再计算”而是“边解压边计算”有些场景下你需要的并不是把解压后的文件落盘而是希望解压内容直接参与统计或过滤。比如zip里放的是订单JSON下游想知道当天订单总量那完全没必要先把JSON全部写出来再读一遍。如果要省掉“解压后落盘再读”两步MapReduce就不太合适更适合用Spark。Spark的RDD或DataFrame允许你在一个map或flatMap算子内部打开HDFS文件自行完成zip的流式解析对解析出来的每行数据进行业务处理。这样数据只经过一次网络读取CPU解压完直接进业务逻辑IO开销省掉一大截。3.2 Spark解压的两种常见写法第一种写法简单粗暴把zip路径列表做成一个RDD然后用flatMap在executor端打开每个包逐行解析。示例逻辑如下val zipFiles spark.sparkContext.binaryFiles(/data/raw/2024/06/*.zip) zipFiles.flatMap { case (path, portableDataStream) val zis new ZipInputStream(portableDataStream.open()) val entries Iterator.continually(zis.getNextEntry()) .takeWhile(_ ! null) .filterNot(_.isDirectory) .flatMap { _ val reader new BufferedReader(new InputStreamReader(zis, StandardCharsets.UTF_8)) Iterator.continually(reader.readLine()).takeWhile(_ ! null) } entries }.filter(line ...) // 业务逻辑第二种写法是用Spark 3.x的binaryFile格式val df spark.read.format(binaryFile) .option(pathGlobFilter, *.zip) .load(/data/raw/2024/06/) df.rdd.flatMap { row val bytes row.getAs[Array[Byte]](content) parseZipFromBytes(bytes).iterator }binaryFile会把整个zip包的内容以字节数组形式加载到executor内存。zip包不太大单包小于单个executor内存时很好用一旦包很大你得改用binaryRecords或者直接用Hadoop API在executor里按块读取。3.3 边解压边计算的性能收益与风险性能收益主要体现在少写少读。假设一个zip包有2GB压缩数据解压出来是6GB原始文件。如果先落盘再计算落地6GB然后读取再加后续计算读取总共至少12GB的HDFS读写边解压边计算只有2GB的HDFS读带宽差异一目了然。风险也集中在内存上。因为zip解压后的内容是流式生成的如果你在flatMap里不小心把所有解析结果collect到一个大集合再返回内存直接爆。务必使用Iterator式的惰性求值也别把大量entry的引用存下来。我踩过一次坑解压几十个ticket文件后把所有数组拼在一个List里结果一个executor直接OOM。后来改成flatMap里逐行处理内存占用立刻降下来了。4. 常见问题与排查技巧实录4.1 报“invalid local file header signature”错误如果zip包在HDFS上明明存在但解压过程中报java.util.zip.ZipException: invalid local file header signature大概率不是代码问题而是zip包本身不完整或已损坏。这种包常见于上传中断或压缩工具版本不一致。处理办法分两层先拿到包在HDFS上的checksum值和源文件做对比确认包是否完整其次在解压代码里捕获ZipException单独把损坏包路径输出到一个die目录并触发告警。不要因为一个坏包拖垮整个批次。4.2 解压任务进度看似卡住在YARN上跑长时间的map如果map既不输出数据也不汇报进度任务超过默认超时时间就会被杀掉。这个坑非常隐蔽。因为map正在解压一个大文件一个entry可能解压几分钟而这段期间没有任何数据被写入contextMR框架就认为任务已经僵死。解法就是我前面说的每处理完一个entry必须调用context.progress()或者context.setStatus()。这两个方法会把“任务仍在活动”的信号上报给ApplicationMaster即使几分钟产出一批任务依然安全。4.3 解压速度奇慢CPU占用上不去单个zip包大到几GB时一个map从头读到尾跑上个把小时很正常。常见瓶颈是任务并发度太低map数量等于zip包数量而zip包分布不平均导致某些节点忙疯、另一些节点闲死。我的排查思路先看数据分布用hadoop fs -du -h看各包大小再看map任务数确认是不是只有一个map在跑一个大包。如果是可以先用distcp或文件切分工具把超大zip按业务逻辑拆成多个小包或者直接在MapReduce之外加一层Spark的repartition提高并行度。另一种情况是CPU利用率不高往往是BufferedInputStream缓冲区太小默认8KB对zip解压来说偏保守我把缓冲区提高到256KB后吞吐有明显提升。4.4 中文文件名解压后变成乱码zip格式的entry名在打包时用的编码跟系统区域设置有关。很多Windows上的压缩工具默认用GBK编码写entry名而Java的ZipInputStream默认按UTF-8解析于是文件名里的中文就变成乱码甚至写入HDFS时直接报非法字符错误。处理方式是手动指定解析编码。如果确认包内entry是GBK编码可以把ZipInputStream包一层自定义的InputStreamReader或者使用commons-compress里的ZipArchiveInputStream它对编码的容忍度好一些。更稳妥的办法是提前和上游约定统一用UTF-8编码做压缩省掉这个历史包袱。4.5 输出目录文件数量爆炸一个zip包动辄包含几万个entry如果全部直接解压成小文件写到HDFS上NameNode的压力会剧增。HDFS本身擅长大文件小文件多了以后元数据内存占用会成倍上涨后续任务扫描目录也会变慢。解决思路有三个方向一是如果业务允许把多个entry合并写成一个SequenceFile或Avro文件保留key为entry路径二是把大量同类小entry经Spark处理后写时做coalesce控制输出文件数三是如果纯粹归档需求考虑直接把HDFS上的小文件定期用Archive机制合并。具体选哪个取决于下游用什么方式消费数据。下面把几个高频问题整理成速查表现象可能原因快速排查与处理ZipException压缩包损坏对比checksum坏包单独隔离并告警map任务被杀长时间未汇报进度在entry级别调用context.progress()解压速度慢并发度低或缓冲区小拆包提升并行度调整缓冲区大小中文文件名乱码GBK/UTF-8编码不一致手动指定entry名编码推动上游统一UTF-8输出小文件过多每entry单文件转存SequenceFile/Avro或做合并5. 参数调优与生产环境扩展建议5.1 YARN资源与任务并行度调优跑MapReduce解压任务时我会重点关注三个参数mapreduce.map.memory.mb、mapreduce.map.java.opts和mapreduce.map.cpu.vcores。zip解压是CPU密集型的map的内存需求反而不高主要在于打开HDFS流和创建输出流时的临时开销。一般情况下我给单个map分配2GB内存堆大小配1GBvcores配2个就能比较从容地处理1GB左右压缩包。包更大或并发任务更多时再往上加内存但不要无脑分配8GB那会导致YARN队列资源被快速占满其他任务等着排队。如果想在一批zip包上尽快解压完关键不在于调大单map内存而在于让map任务数等于集群可用槽位数。控制输入路径里zip包的数量和大小分布比盲目加大资源更有效。5.2 数据校验与解压结果确认生产环境不能解压完就当成功必须加校验环节。我在Mapper写完输出之后会给每个输出目录写一个_SUCCESS标记文件里面记录来源zip包的checksum、解压时间、文件数统计等元信息。下游任务启动前先去检查这个_SUCCESS文件存不存在存在才继续处理。另外如果下游对数据完整性有硬性要求可以在解压后对输出文件目录做一次fs -du的字节数校验预估解压后大小与源码包中记录的文件大小是否匹配。虽然zip解压后的总大小不能从中央目录直接拿到精确值但可以通过解析每个entry的size字段汇总代码里加一个累加器即可这点成本非常低。5.3 从单次解压到可复用解压组件如果团队不止一个业务方在解压zip建议把解压逻辑封装成标准组件而不是每个管道各写一遍。我之前就把这个Mapper和InputFormat封装进一个公共JAR包对外只留配置项输入目录、输出目录、是否保留包内目录结构、是否覆盖目标文件、文件名冲突规则。这样业务方只需要在调度平台上传一个配置就能跑不需要关心底层MapReduce怎么实现。后续如果发现新的编码问题或压缩工具兼容问题只改组件代码重新打包即可所有业务一次生效。封装之后的扩展场景也很多。比如增量解压只处理指定时间之后进入的zip包又比如与告警系统联动当解压失败率超过阈值时自动发出通知。这些在组件化之前很难做到统一组件化之后全部变成配置项的事情。最后说点个人体会我个人的习惯是能不在本地机器上碰数据就坚决不碰。HDFS上的zip解压看似是个不起眼的小需求但把它做扎实了能省下非常多的运维精力。踩过几次坑之后我总结出一条铁律每次解压任务结束一定要留一个可追踪的标记文件有了它下游任务怎么排都不会乱。后续如果想继续扩展还可以在解压的同时做数据质量探针把空文件、异常字符直接标记出来让问题在上游暴露而不是等下游分析时才发现。这套逻辑放到任何数据格式上都通用看的人可以按自己实际场景调整。
阅读完成 · 觉得有帮助？
