如何按天合并HDFS中拆分的小型ORC日志文件（Java实现）

针对你这种按天拆分的日志ORC文件，需要在保留天级独立性的前提下合并目录内小文件的场景，我整理了几个最实用的Java实现方案，结合你的需求帮你分析优劣：

一、先说说你提到的OrcFileMergeOperator是否适用 #

OrcFileMergeOperator是Hive执行引擎内部的一个算子，主要用于Hive执行查询计划时自动合并小ORC文件（比如开启hive.merge.orcfiles配置后）。但它并不适合作为独立Java程序的核心组件：

• 它强依赖Hive的执行环境（比如HiveConf、TaskContext等），脱离Hive查询流程单独使用会非常繁琐；
• 它的设计是为Hive的批量查询优化服务，而非独立的文件合并工具。

所以不建议直接基于它来实现你的需求。

二、推荐的三种Java实现方案 #

1. 轻量方案：直接使用ORC官方Java库合并 #

这种方案无需依赖Hive集群，只引入ORC的Java依赖即可，适合独立的合并工具开发。

步骤说明：

• 遍历目标天目录下的所有ORC小文件；
• 校验所有文件的Schema一致性（日志文件通常Schema统一，可跳过校验但建议保留）；
• 创建ORC Writer，将所有小文件的内容批量写入到一个大ORC文件；
• 写入完成后，可删除原小文件或归档到指定目录。

代码示例：

首先引入Maven依赖：

<dependency>
    <groupId>org.apache.orc</groupId>
    <artifactId>orc-core</artifactId>
    <version>1.8.4</version> <!-- 选择对应集群的ORC版本 -->
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.3.4</version>
</dependency>

核心合并逻辑：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.orc.OrcFile;
import org.apache.orc.Reader;
import org.apache.orc.RecordReader;
import org.apache.orc.TypeDescription;
import org.apache.orc.Writer;

import java.util.ArrayList;
import java.util.List;

public class OrcDailyMerger {
    public static void mergeDayDirectory(Configuration conf, String dayDirPath, String mergedFilePath) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        Path dayDir = new Path(dayDirPath);
        List<Path> orcFiles = listOrcFiles(fs, dayDir);

        // 读取第一个文件的Schema作为目标Schema
        Reader firstReader = OrcFile.createReader(orcFiles.get(0), OrcFile.readerOptions(conf));
        TypeDescription schema = firstReader.getSchema();

        // 创建合并后的ORC Writer
        Writer writer = OrcFile.createWriter(new Path(mergedFilePath),
                OrcFile.writerOptions(conf)
                        .setSchema(schema)
                        .setCompression(OrcFile.CompressionKind.ZLIB)); // 匹配原文件的压缩格式

        // 遍历所有小文件，写入到Writer
        for (Path file : orcFiles) {
            Reader reader = OrcFile.createReader(file, OrcFile.readerOptions(conf));
            RecordReader recordReader = reader.rows();
            Object row;
            while (recordReader.hasNext()) {
                row = recordReader.next(null);
                writer.addRow(row);
            }
            recordReader.close();
        }

        writer.close();
        // 可选：删除原小文件
        for (Path file : orcFiles) {
            fs.delete(file, false);
        }
    }

    // 辅助方法：列出目录下所有ORC文件
    private static List<Path> listOrcFiles(FileSystem fs, Path dir) throws Exception {
        List<Path> orcFiles = new ArrayList<>();
        if (fs.exists(dir) && fs.isDirectory(dir)) {
            for (org.apache.hadoop.fs.FileStatus status : fs.listStatus(dir)) {
                Path filePath = status.getPath();
                if (filePath.getName().endsWith(".orc")) {
                    orcFiles.add(filePath);
                }
            }
        }
        return orcFiles;
    }
}

2. 便捷方案：利用Hive Java API执行分区合并SQL #

如果你的环境已经部署了Hive集群，这是最省心的方案——直接通过Hive的SQL来合并指定天的分区，Hive会自动处理ORC文件的合并逻辑（需开启相关优化配置）。

核心思路：

执行INSERT OVERWRITE语句覆盖目标分区，Hive会将该分区下的小文件合并为少量大文件（可通过hive.merge.orcfiles=true、hive.merge.size.per.task等配置调整合并后的文件大小）。

代码示例（JDBC方式）：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;

public class HiveOrcMerger {
    public static void mergeHivePartition(String dt) throws Exception {
        // 加载Hive JDBC驱动
        Class.forName("org.apache.hive.jdbc.HiveDriver");
        // 建立连接（替换为你的HiveServer2地址、账号信息）
        Connection conn = DriverManager.getConnection("jdbc:hive2://your-hiveserver2:10000/default", "username", "password");
        Statement stmt = conn.createStatement();

        // 执行合并SQL
        String sql = String.format(
            "INSERT OVERWRITE TABLE your_log_table PARTITION (dt='%s') " +
            "SELECT * FROM your_log_table WHERE dt='%s'", dt, dt
        );
        stmt.execute(sql);

        stmt.close();
        conn.close();
    }
}

3. 大数据量方案：基于MapReduce的分布式合并 #

如果单天的ORC文件数量极大、数据量超过单节点处理能力，可以用MapReduce来分布式合并：

• 使用OrcInputFormat读取某天目录下的所有ORC文件；
• Map阶段直接输出读取到的行数据，不做处理；
• Reduce阶段将所有数据写入到一个（或少量）大ORC文件（通过设置mapreduce.job.reduces=1控制输出文件数量）。

这种方案适合超大规模的日志合并，利用Hadoop集群的分布式能力处理。

三、方案选择建议 #

• 如果是独立工具，无Hive集群依赖：选「ORC官方Java库」方案，轻量灵活；
• 已有Hive集群，希望快速实现：选「Hive Java API执行SQL」方案，无需处理ORC底层细节，省心高效；
• 单天数据量极大，需要分布式处理：选「MapReduce分布式合并」方案，充分利用集群资源。

内容的提问来源于stack exchange，提问作者Chris C