如何按文件名指定顺序向Apache Flink流式加载CSV文件？

你提到的Spark流式读取场景，在Flink里可以通过FileSource API（Flink 1.14+推荐的新文件源）来实现，结合你逐个按顺序移动文件的操作习惯，我们可以通过配置文件枚举逻辑和排序规则来保证处理顺序，下面是具体的实现步骤和代码示例：

1. 核心思路：原子文件移动+可控文件发现 #

和Spark的逻辑类似，你依然可以保持**先将文件原子移动到监控目录（如/data/staging）**的操作流程——同一文件系统内的mv操作是原子性的，Flink只会读取完全写入的文件。接下来通过Flink的FileSource配置，让它按你需要的顺序（比如文件名中的时间戳）处理新发现的文件。

2. 具体代码实现 #

2.1 依赖准备 #

确保你的项目依赖了Flink的文件源和CSV格式包（以Maven为例）：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-formats-csv</artifactId>
    <version>${flink.version}</version>
</dependency>

2.2 构建按文件名排序的流式CSV读取任务 #

下面的代码会监控指定目录，按文件名（比如带时间戳的命名）升序处理新文件，并且保证只处理一次：

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.RowTypeInfo;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.CsvReaderFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;

public class OrderedCsvFileStreaming {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 开启Checkpointing，持久化已处理文件状态，重启后不重复处理
        env.enableCheckpointing(30000); // 30秒一次checkpoint

        // 定义CSV Schema，和你Spark中的schema对应（示例：id(int), name(string), create_time(string)）
        RowTypeInfo rowTypeInfo = new RowTypeInfo(
                Integer.class,
                String.class,
                String.class
        );

        // 构建CSV读取格式，开启表头解析
        CsvReaderFormat<Row> csvFormat = CsvReaderFormat.forRowType(
                rowTypeInfo,
                CsvReaderFormat.DEFAULT_FIELD_DELIMITER,
                CsvReaderFormat.DEFAULT_LINE_DELIMITER
        )
        .setHasHeaderLine(true);

        // 构建FileSource，配置监控目录和自定义文件枚举逻辑
        FileSource<Row> fileSource = FileSource.forRecordStreamFormat(
                csvFormat,
                new Path("/data/staging")
        )
        // 定期扫描目录（10秒一次，可根据你的文件移动频率调整）
        .monitorContinuously(10000)
        // 自定义文件排序：按文件名升序（适配带时间戳的文件名，比如20240520_100000.csv）
        .withFileEnumerator(context -> {
            return new ContinuousFileEnumerator(10000) {
                @Override
                protected java.util.Comparator<Path> getPathComparator() {
                    // 若文件名时间戳格式特殊，可在这里解析时间戳后再排序
                    return (path1, path2) -> path1.getName().compareTo(path2.getName());
                }
            };
        })
        // 排除隐藏文件，避免读取临时文件
        .excludeHiddenFiles()
        .build();

        // 读取文件流并执行业务处理
        env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "Ordered CSV File Source")
                .print(); // 替换为你的实际业务逻辑

        env.execute("Ordered CSV Streaming Job");
    }
}

3. 关键细节说明 #

• 原子文件移动：必须保证文件通过原子操作移动到监控目录（比如Linux的mv命令，同一磁盘分区内是原子的），防止Flink读取未完全写入的文件。
• 自定义排序逻辑：如果你的文件名时间戳格式不是纯字符串顺序（比如带横杠的2024-05-20_10:00:00.csv），可以在比较器里解析时间戳后再按时间排序。
• Checkpointing的必要性：开启Checkpointing后，Flink会记录已处理的文件列表，任务重启后不会重复处理历史文件。
• 扫描间隔调整：如果你的文件移动频率很高，可以缩短monitorContinuously的间隔，让Flink更快发现新文件。

4. 旧版Flink兼容方案（1.13及以下） #

如果还在使用旧版Flink，可以使用readFile API配合FileProcessingMode.PROCESS_CONTINUOUSLY，但需要自定义InputFormat来控制文件顺序，这种方式灵活性较差，推荐尽快升级到新版Flink使用FileSource。

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