代码之家 › 专栏 › 技术社区 › Daniel

目录中的Apache Flink流文件

apache-flink

Daniel · 技术社区 · 6 年前

我想按照给定的顺序将一组csv文件加载到Apache Flink中,例如,根据文件名中的命名方案确定,其中可能包含一些时间戳信息。

在Apache Spark中,只要将文件移动到特定目录(例如/数据/暂存目录),我就可以将文件流式传输到数据集,原子文件移动如下

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

然后,我会按照给定的顺序(例如使用bash脚本)将文件逐个移动到该临时目录。

如何使用Apache Flink实现同样的效果?

1 回复 | 直到 6 年前

kkrugler 6 年前

这并不完全是相同的用例,但我们必须在流媒体作业中执行类似的操作(文件是 HDF5 不 CSV )所以我写了一个 RichSourceFunction 它知道如何按正确的顺序遍历文件,并将文件路径(在S3中)作为字符串记录发出。然后是下游 FlatMapFunction 解析文件并发出实际行。

推荐文章

Ellen Spertus · 使用Scala访问HBase的推荐方法

6 年前

skrshn · JAVA尝试在Flink中创建Kafka使用者时出现lang.NoClassDefFoundError(CheckpointedRestoring)

6 年前

Daniel · 目录中的Apache Flink流文件

6 年前

Christos Hadjinikolis · Flink:执行管道java时出错。util。同时发生的TimeoutException:期货在[10000毫秒]后超时

6 年前

Mustafa · Apache Flink:自定义InputFormat仅以1的并行度运行

7 年前

Soheil Pourbafrani · Apache Flink-org的类文件。阿帕奇。弗林克。流式处理。api。斯卡拉。找不到数据流

7 年前

Jicaar · 一个卡夫卡制作人可以被多个接收器安全使用吗?

7 年前

Christos Hadjinikolis · Flink JobExecutionException:akka。客户超时

7 年前

Christos Hadjinikolis · Flink IOException:网络缓冲区数量不足

7 年前

Brutal_JL · flink ReduceFunction的默认触发器是什么?

7 年前