来自目录的 Apache Flink 流式处理文件

Question

我想以给定的顺序将一堆 csv 文件加载到 Apache Flink 中，例如由文件名中的命名方案确定，其中可能包含一些时间戳信息。

在 Apache Spark 中，只要将文件移动到特定目录（例如 /data/staging），我就可以将文件流式传输到数据集，原子文件移动如下

Dataset<Row> fileStreamDf = spark.readStream()
            .option("header", true)
            .schema(schema)
            .csv("/data/staging")

然后我会按照给定的顺序一个一个地移动文件，例如使用 bash 脚本，到该暂存目录。

如何使用 Apache Flink 实现相同的目标？

Answer 1

这不是完全相同的用例，但我们必须在流式作业中做类似的事情（文件是 HDF5 而不是 CSV）。所以我写了一个RichSourceFunction，它知道如何以正确的顺序迭代文件，并将文件路径（这些在 S3 中）作为字符串记录发出。然后下游FlatMapFunction 解析文件并发出实际的行。