5.7、Flink流处理（Stream API）- Connectors（外部连接器）之 Hadoop FileSystem Connector

ゝ一世哀愁。 2023-06-25 06:15 13阅读 0赞

> 自 Flink 1.9 起，BucketingSink 已经被弃用，并将在后续版本中移除。请使用 StreamingFileSink。

这个连接器提供了一个接收器，可以将分区的文件写到Hadoop文件系统支持的任何文件系统中。要使用此连接器，请在项目中添加以下依赖项：

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-filesystem_2.11</artifactId>
      <version>1.9.0</version>
    </dependency>

> 请注意，流连接器目前不是集群 lib 的一部分。有关如何将程序与库打包以供集群执行的信息，请参阅：[Flink基础之配置Maven依赖、连接、库][Flink_Maven]。

## Bucketing File Sink ##

过滤行为和写入可以配置，但我们将在稍后进行讨论。这是你如何可以创建一个 bucketing 接收器，默认情况下，接收器通过时间来拆分文件：

val input: DataStream[String] = ...
    
    input.addSink(new BucketingSink[String]("/base/path"))

惟一需要的参数是存储bucket的基本路径。可以通过指定自定义bucketer、写入器和批大小来进一步配置接收器。

默认情况下，当元素到达时，bucketing sink 将按照当前系统时间进行分割，并使用datetime模式“yyyy-MM-dd- HH”来命名 bucket。此模式将与当前系统时间和 JVM 的默认时区一起传递给 DateTimeFormatter，以形成 bucket 路径。用户还可以为 bucketer 指定一个时区来格式化 bucket 路径。每当遇到新的日期时，就会创建一个新的 bucket。例如，如果您有一个包含分钟作为最细粒度的模式，那么每分钟您都会得到一个新桶。

当桶变得不活动时，打开的部分文件将被刷新并关闭。当一个桶最近没有被写入时，它被认为是不活动的。默认情况下，接收器每分钟检查不活动的桶，并关闭超过一分钟没有写入的桶。可以在 BucketingSink 上使用 setInactiveBucketCheckInterval()和 setInactiveBucketThreshold()配置此行为。

您还可以通过在 BucketingSink 上使用 setBucketer()来指定自定义 bucketer。如果需要，bucketer 可以使用元素或元组的属性来确定 bucket 目录。

默认的写入器是 StringWriter。这将对传入的元素调用 toString()，并将它们写入用换行符分隔的部分文件。要指定自定义写入器，请在 BucketingSink 上使用 setWriter()。如果您想编写 Hadoop sequencefile 文件，您可以使用提供的 SequenceFileWriter，它也可以配置使用压缩。

有两个配置选项可以指定何时关闭一个文件，何时启动一个新的文件：

*  通过设置批大小(默认的部分文件大小为384 MB)
 *  通过设置一个随时间间隔的批滚(默认的随时间间隔是Long.MAX\_VALUE)

当满足这两个条件之一时，将启动一个新的部件文件。

// the SequenceFileWriter only works with Flink Tuples
    import org.apache.flink.api.java.tuple.Tuple2
    val input: DataStream[Tuple2[A, B]] = ... 
    
    val sink = new BucketingSink[Tuple2[IntWritable, Text]]("/base/path")
    sink.setBucketer(new DateTimeBucketer("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)

这将创建一个接收器，写入桶文件遵循这个模式：

/base/path/{date-time}/part-{parallel-task}-{count}

其中 date-time 是我们从 date/time 格式获得的字符串，parallel-task 是并行接收实例的索引，count 是由于批大小或批处理时间间隔而创建的部分文件的运行数量。

原文地址：[https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/filesystem\_sink.html][https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_connectors_filesystem_sink.html]

[Flink_Maven]: https://blog.csdn.net/zpf_940810653842/article/details/99622787
[https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_connectors_filesystem_sink.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/filesystem_sink.html