如何基于字段将CSV数据加载到不同的Hadoop HDFS目录中

BufferedReader的适用性分析 #

BufferedReader本身不是错误选型，如果你处理的是中小规模（单文件GB级以下、无超高并发写入需求）的CSV文件，该方案完全可用，你遇到的落地障碍大概率是用法问题而非方案本身的问题，常见的容易踩的坑包括：

• 直接用split(",")分割CSV行，没有处理带逗号的带引号字段、转义符，导致字段错位
• 流操作没有使用try-with-resources语法自动释放资源，导致文件句柄泄漏、HDFS连接残留
• HDFS输出流没有正确flush/close，导致写入的HDFS文件块损坏、内容缺失

核心实现的参考代码如下：

// 需提前引入hadoop-common、hadoop-hdfs依赖，版本和集群版本对齐
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.*;
import java.net.URI;
import java.nio.charset.StandardCharsets;

public class CsvToHdfs {
    public static void main(String[] args) throws Exception {
        // 初始化HDFS连接
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://your-namenode-ip:9000"), conf);
        String localCsvPath = "/path/to/your/local.csv";

        try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(localCsvPath), StandardCharsets.UTF_8))) {
            String line;
            // 跳过CSV表头（如果无表头可删除此行）
            br.readLine();
            while ((line = br.readLine()) != null) {
                // 复杂CSV建议替换为opencsv/commons-csv的解析逻辑，避免格式错误
                String[] fields = line.split(",");
                // 替换为你实际的年份字段下标
                String year = fields[3];
                Path hdfsTargetPath = new Path("/data/root/year=" + year + "/part.csv");
                // 支持追加写入，避免同分区生成大量小文件
                try (FSDataOutputStream out = fs.exists(hdfsTargetPath) ? fs.append(hdfsTargetPath) : fs.create(hdfsTargetPath)) {
                    out.write(line.getBytes(StandardCharsets.UTF_8));
                    out.write("\n".getBytes(StandardCharsets.UTF_8));
                }
            }
        } finally {
            fs.close();
        }
    }
}

更优的替代方案 #

根据你的场景差异，可以选择更适配的方案：

• 如果CSV格式复杂，存在带引号的字段、字段内换行、转义符等规则：优先用opencsv、commons-csv等成熟CSV解析库替换原生的split逻辑，性能和BufferedReader差异极小，且能避免90%以上的格式解析错误
• 如果是处理10GB以上的大文件、多批次批量任务：可以直接使用Hadoop MapReduce或者Spark框架实现，不需要手写IO逻辑，框架自动处理分片、容错、负载均衡，只需要核心的抽取年份字段、路由输出目录的逻辑即可，Java语言完全支持两个框架的开发
• 如果要求极致的IO性能、低资源开销：可以用BufferedInputStream配合NIO的FileChannel实现零拷贝读取，比BufferedReader的IO效率高30%以上，适合大文件处理场景

选型建议 #

• 一次性小批量任务、CSV格式简单：BufferedReader+简单解析的组合完全够用，不需要引入额外框架
• 生产级定时任务、数据量波动大：优先用Spark SQL实现，几行代码就能完成按年份字段分区写入HDFS的逻辑，可维护性和稳定性远高于手写IO代码
• 所有场景都建议开启HDFS的追加写入权限，避免同一年份的多批次数据生成大量小文件，影响HDFS性能

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