在上一章节已经翻译了Flink卡夫卡连接器,但由于HDFS作为大多数研究大数据者日常用到的,此章节并添加翻译HDFS的连接器。
此连接器提供了一个水池,将分区文件写入Hadoop文件系统支持的任何文件系统。要使用此连接器,请将以下依赖项添加到您的项目中:
& lt; dependency>
,& lt; groupId>表示。flink & lt;/groupId>
,& lt; artifactId>flink-connector-filesystem_2.10 & lt;/artifactId>
,& lt; version>1.2.0 & lt;/version>
& lt;/dependency>
请注意,流连接器当前不是二进制分发的一部分。有关如何将程序与程序库打包以进行集群执行的信息,请参阅此处。
<强>折叠文件接收器( <强>)
可以配置压力行为以及写入操作,但我们稍后将会介绍。这是你如何创建一个耐心的病人,默认情况下,它会收敛到按时间分割的滚动文件:
& lt; Java代码在
DataStream数据& lt; String>输入=?
输入。addSink(新BucketingSink & lt; String>(“/基地/路径”)),
唯一必需的参数是存储桶的基本路径。可以通过指定自定义桶,写入器和批量大小来进一步配置接收器。
默认情况下,当元素到达时,当前的系统时间将会降级,并使用日期时间模式“yyyy-MM-dd-HH”命名这些存储区。此模式将传递给具有当前系统时间的SimpleDateFormat以形成存储桶路径。每当遇到新的日期时,都会创建一个新的桶,例如,如果您有一个包含分钟作为最细粒度的模式,您将每分钟获得一个新的桶。每个桶本身是一个包含几个零件文件的目录:每个并行实例的接收器将创建自己的零件文件,当零件文件变得太大时,槽也将在其他文件旁边创建一个新的零件文件。当桶变得不活动时,打开的零件文件将被刷新并关闭。当最近没有写入时,桶被视为不活动。默认情况下,接收器每分钟检查不活动的桶,并关闭一分钟内未写入的任何桶。可以在BucketingSink上使用setInactiveBucketCheckInterval()和setInactiveBucketThreshold()配置此行为。
您也可以使用BucketingSink上的setBucketer()指定自定义桶。如果需要,桶可以使用元素或元组的属性来确定桶目录。
默认的作者是StringWriter。这将调用toString()对传入的元素,并将它们写入部分文件,用换行符分隔。要在BucketingSink上指定一个自定义的作者,请使用setWriter()。如果要编写Hadoop SequenceFiles,可以使用提供的SequenceFileWriter,它也可以配置为使用压缩。
最后一个配置选项是批量大小。这指定何时应该关闭零件文件并启动一个新的零件。(默认部分文件大小为384 MB)。
例:
& lt; Java代码在
DataStream数据& lt; Tuple2 & lt; IntWritable,文本在祝辞输入=?
sinketingSink & lt; String>水槽=new BucketingSink & lt; String>(“/基地/路径”);
下沉。setBucketer(新DateTimeBucketer & lt; String> (“yyyy-MM-dd-HHmm”)),
下沉。setWriter(新SequenceFileWriter & lt; IntWritable Text> ());
下沉。setBatchSize (1024 * 1024 * 400);//这是400 MB,
input.addSink(下沉);
这将创建一个写入到遵循此模式的桶文件的接收器:
//道路/{日期-时间}/基础部分——{并行任务}{数}
其中日期-时间是从日期/时间格式获取的字符串、并行任务是并行接收器实例的索引,数是由于批量大小而创建的部分文件的运行数。
