如何使用Apache Flink DataSet API批量读取目录下多文本文件？

嘿，我来帮你搞定这个问题！你说得对，循环为每个文件单独启动作业绝对不是最优解——Flink作业启动本身就有开销，批量处理才是正道。下面给你几个实用的方案，你可以根据自己的需求来选：

1. 直接读取整个目录（最简单的方式） #

其实readTextFile()本身就支持传入目录路径，而不仅仅是单个文件路径！Flink会自动读取该目录下所有非隐藏的文本文件（默认不递归读取子目录）。示例代码：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<String> allLines = env.readTextFile("/your/directory/path");
// 后续处理逻辑
allLines.map(new YourMapFunction()).print();

如果需要递归读取子目录里的文件，你可以用更底层的readFile()方法配合TextInputFormat：

TextInputFormat format = new TextInputFormat(new Path("/your/directory/path"));
// 开启递归枚举子目录
format.setNestedFileEnumeration(true);
DataSet<String> allLines = env.readFile(format, "/your/directory/path");

2. 读取文件时获取文件元数据（如文件名、路径） #

如果你的处理逻辑需要知道每条记录来自哪个文件，这个方案更适合。通过RichMapFunction可以获取输入拆分的信息，从而拿到文件路径：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TextInputFormat format = new TextInputFormat(new Path("/your/directory/path"));
format.setNestedFileEnumeration(true);

DataSet<Tuple2<String, String>> linesWithFileName = env.readFile(format, "/your/directory/path")
    .map(new RichMapFunction<String, Tuple2<String, String>>() {
        @Override
        public Tuple2<String, String> map(String value) throws Exception {
            // 获取当前输入拆分对应的文件路径
            FileInputSplit split = (FileInputSplit) getRuntimeContext().getInputSplit();
            String filePath = split.getPath().toString();
            return new Tuple2<>(filePath, value);
        }
    });

// 后续可以按文件分组处理，或者做其他逻辑
linesWithFileName.groupBy(0).reduce(new YourReduceFunction()).print();

3. 自定义文件过滤与逐个处理 #

如果需要对要读取的文件做更细粒度的控制（比如只处理特定后缀的文件，或者排除某些文件），可以先遍历目录收集符合条件的文件路径，再逐个读取：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Path dirPath = new Path("/your/directory/path");
FileSystem fs = dirPath.getFileSystem();
FileStatus[] fileStatuses = fs.listStatus(dirPath);

// 收集所有符合条件的文件路径（这里示例只取.txt文件）
List<String> filePaths = new ArrayList<>();
for (FileStatus status : fileStatuses) {
    if (!status.isDir() && status.getPath().getName().endsWith(".txt")) {
        filePaths.add(status.getPath().toString());
    }
}

// 将文件路径转为DataSet，然后逐个读取文件内容
DataSet<String> allLines = env.fromCollection(filePaths)
    .flatMap(new FlatMapFunction<String, String>() {
        @Override
        public void flatMap(String filePath, Collector<String> out) throws Exception {
            // 读取单个文件的每一行
            try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    out.collect(line);
                    // 如果需要带上文件名，也可以输出Tuple2<String, String>
                    // out.collect(new Tuple2<>(filePath, line));
                }
            }
        }
    });

// 后续处理逻辑
allLines.print();

为什么不推荐循环启动作业？ #

正如你所想的，每次启动Flink作业都需要申请资源、初始化环境、分发代码，这些开销累加起来会非常大，尤其是文件数量多的时候。上面的方案都是在同一个作业内完成所有文件的处理，能最大化利用集群资源，效率高得多。

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