1.4 Flink HDFS Connector /Flink HDFS連接器

在上一章節已經翻譯了Flink Kafka Connector，但由于HDFS作為大多數研究大數據者日常用到的，此章節并添加翻譯HDFS的連接器。

此連接器提供了一個Sink，將分區文件寫入Hadoop FileSystem支持的任何文件系統。要使用此連接器，請將以下依賴項添加到您的項目中：

<dependency>

  <groupId> org.apache.flink </groupId>

  <artifactId> flink-connector-filesystem_2.10 </artifactId>

  <version> 1.2.0 </version>

</dependency>

請注意，流連接器當前不是二進制分發的一部分。有關如何將程序與程序庫打包以進行集群執行的信息，請參閱此處。

折疊文件接收器（Bucketing File Sink）

可以配置壓力行為以及寫入操作，但我們稍后將會介紹。這是你如何創建一個耐心的病人，默認情況下，它會收斂到按時間分割的滾動文件：

<Java代碼>

DataStream <String> input = ...;

input.addSink（new BucketingSink <String>（“/base/path”））;

唯一必需的參數是存儲桶的基本路徑。可以通過指定自定義bucketer，寫入器和批量大小來進一步配置接收器。

默認情況下，當元素到達時，當前的系統時間將會降級，并使用日期時間模式“yyyy-MM-dd-HH”命名這些存儲區。此模式將傳遞給具有當前系統時間的SimpleDateFormat以形成存儲桶路徑。每當遇到新的日期時，都會創建一個新的桶。例如，如果您有一個包含分鐘作為最細粒度的模式，您將每分鐘獲得一個新的桶。每個桶本身是一個包含幾個零件文件的目錄：每個并行實例的接收器將創建自己的零件文件，當零件文件變得太大時，槽也將在其他文件旁邊創建一個新的零件文件。當桶變得不活動時，打開的零件文件將被刷新并關閉。當最近沒有寫入時，桶被視為不活動。默認情況下，接收器每分鐘檢查不活動的桶，并關閉一分鐘內未寫入的任何桶。可以在BucketingSink上使用setInactiveBucketCheckInterval（）和setInactiveBucketThreshold（）配置此行為。

您也可以使用BucketingSink上的setBucketer（）指定自定義bucketer。如果需要，bucketer可以使用元素或元組的屬性來確定bucket目錄。

默認的作者是StringWriter。這將調用toString（）對傳入的元素，并將它們寫入部分文件，用換行符分隔。要在BucketingSink上指定一個自定義的作者，請使用setWriter（）。如果要編寫Hadoop SequenceFiles，可以使用提供的SequenceFileWriter，它也可以配置為使用壓縮。

最后一個配置選項是批量大小。這指定何時應該關閉零件文件并啟動一個新的零件。 （默認部分文件大小為384 MB）。

例：

<Java代碼>

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

sinketingSink <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）; //這是400 MB，

input.addSink（sink）;

這將創建一個寫入到遵循此模式的桶文件的接收器：

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

其中date-time是從日期/時間格式獲取的字符串，parallel-task是并行接收器實例的索引，count是由于批量大小而創建的部分文件的運行數。