这个Connector提供了一个sink来写分区文件到任何Hadoop FileSystem支持的任何文件系统中，为了使用这个Connector，请将下面的依赖添加到你的工程中:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-filesystem_2.10</artifactId>
<version>1.3.0</version>
</dependency>
注意:streaming connectors目前还不是二进制发布包的一部分，请参考此处来了解如何在分布式执行中关联到这些connectors。
Flink Sink分桶
分桶行为跟写行为都是可以配置的，这个后面我们会讲到，你可以通过默认配置来创建一个分桶的sink,将数据sink到以时间作为划分的滚动文件中:
Java 代码:
DataStream<String> input = ...;
input.addSink(new BucketingSink<String>("/base/path"));
Scala 代码:
val input: DataStream[String] = ...
input.addSink(new BucketingSink[String]("/base/path"))
这里唯一需要参数是这些分桶所要存储的目录地址，sink还可以通过配置一个自定义的bucketer、writer和批大小来进一步配置。
默认情况下分桶sink是通过元素到达的系统时间来进行切分的，并用"yyyy-MM-dd HH"的时间格式来命名桶，这个时间格式与当前的系统时间传入SimpleDateFormat来形成一个桶的路径，当遇到一个新的时间后就会创建一个新的桶。例如:如果你有一个以分钟作为最细粒度的模式，那么你将每分钟获得一个新的分桶。每个分桶本身是一个包含若干分区文件的目录，每个并行的sink实例会创建它自己的分区文件，当分区文件过大时，sink会紧接着其它分区文件创建一个新的分区文件。当一个桶变成非活跃状态时，打开的文件会被刷新和关闭，当一个桶不再被写入时，会被认为是非活跃的。默认情况下，sink会每分钟检查一遍是否非活跃，并关闭超过一分钟没有数据写入的分桶，这种行为可以通过在BucketingSink的
setInactiveBucketCheckInterval() 和 setInactiveBucketThreshold()来配置。
你可以在BucketingSink中使用setBucketer()来指定一个自定义的bucketer,如果需要，bucketer可以使用元素或者元组的属性来决定bucketer的目录。
默认的writer是StringWriter，这个writer会调用到达的元素的toString()方法，将数据以新的行作为划分写入到分区文件中。你可以在BucketingSink中使用setWriter()来指定一个自定义的writer，如果你想写到Hadoop SequenceFiles，你可以只用预定义的SequenceFileWriter，这个writer还可以指定压缩格式。
最后的配置项是批大小，这个配置指定了一个分区文件何时需要被关闭、新的分区文件开始。(默认的分区文件大小是384MB)
例如:
Java 代码:
DataStream<Tuple2<IntWritable,Text>> input = ...;
BucketingSink<String> sink = new BucketingSink<String>("/base/path");
sink.setBucketer(new DateTimeBucketer<String>("yyyy-MM-dd--HHmm"));
sink.setWriter(new SequenceFileWriter<IntWritable, Text>());
sink.setBatchSize(1024 * 1024 * 400); // this is 400 MB,
input.addSink(sink);
Scala 代码:
val input: DataStream[Tuple2[IntWritable, Text]] = ...
val sink = new BucketingSink[String]("/base/path")
sink.setBucketer(new DateTimeBucketer[String]("yyyy-MM-dd--HHmm"))
sink.setWriter(new SequenceFileWriter[IntWritable, Text]())
sink.setBatchSize(1024 * 1024 * 400) // this is 400 MB,
input.addSink(sink)
这个例子会创建一个按下面的模式来写数据到分桶文件的sink:
/base/path/{date-time}/part-{parallel-task}-{count}
这里date-time是我们从date/time模式中获取的字符串，parallel-task是并行sink实例的索引，count是分区文件的运行编号，这个运行编号是由于分区文件的批大小导致的。
网友评论