关于flink溪流水槽至hdfs

Question

我正在编写一个flink代码，其中我正在从本地系统读取一个文件，并使用“writeUsingOutputFormat”将其写入数据库。

现在我的要求是写入hdfs而不是数据库。

你能帮我在Flink怎么办吗。

注意：hdfs已启动并在本地计算机上运行。

Answer 1

在这一点上，较新的流文件接收器可能是比Bucketing接收器更好的选择。此描述来自Flink 1.6发行说明（请注意，Flink 1.7中添加了对S3的支持）：

新的StreamingFileSink是一个用于写入文件系统的一次性接收器，它利用了从以前的BucketingSink中获得的知识。通过将接收器与Flink的检查点机制集成来支持精确一次。新的接收器建立在Flink自己的文件系统抽象的基础上，它支持本地文件系统和HDFS，计划在不久的将来支持S3[现在包含在Flink 1.7中]。它公开了可插拔文件滚动和屈曲策略。除了按行编码格式之外，新的StreamingFileSink还支持拼花。可以使用公开的API轻松添加ORC等其他批量编码格式。

Answer 2

Flink提供了HDFS连接器，可用于将数据写入Hadoop文件系统支持的任何文件系统。

提供的接收器是一个缓冲接收器，它将数据流分区到包含滚动文件的文件夹中。屈曲行为，以及写入，可以配置参数，如批次大小和批次滚动时间间隔

Flink文件给出了以下例子-

DataStream<Tuple2<IntWritable,Text>> input = ...;

BucketingSink<String> sink = new BucketingSink<String>("/base/path");
sink.setBucketer(new DateTimeBucketer<String>("yyyy-MM-dd--HHmm", ZoneId.of("America/Los_Angeles")));
sink.setWriter(new SequenceFileWriter<IntWritable, Text>());
sink.setBatchSize(1024 * 1024 * 400); // this is 400 MB,
sink.setBatchRolloverInterval(20 * 60 * 1000); // this is 20 mins

input.addSink(sink);