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

創(chuàng)新互聯(lián)是專業(yè)的宜黃網(wǎng)站建設(shè)公司，宜黃接單;提供成都網(wǎng)站設(shè)計、成都做網(wǎng)站,網(wǎng)頁設(shè)計,網(wǎng)站設(shè)計,建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行宜黃網(wǎng)站開發(fā)網(wǎng)頁制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊,希望更多企業(yè)前來合作!

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

<dependency>

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

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

 <version> 1.2.0 </version>

</dependency>

請注意，流連接器當(dāng)前不是二進(jìn)制分發(fā)的一部分。有關(guān)如何將程序與程序庫打包以進(jìn)行集群執(zhí)行的信息，請參閱此處。

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

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

<Java代碼>

DataStream <String> input = ...;

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

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

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

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

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

最后一個配置選項是批量大小。這指定何時應(yīng)該關(guān)閉零件文件并啟動一個新的零件。 （默認(rèn)部分文件大小為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）;

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

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

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

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點(diǎn)與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

名稱欄目：1.4FlinkHDFSConnector/FlinkHDFS連接器-創(chuàng)新互聯(lián)
鏈接分享：http://chinadenli.net/article48/hhehp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供移動網(wǎng)站建設(shè)、品牌網(wǎng)站設(shè)計、域名注冊、響應(yīng)式網(wǎng)站、網(wǎng)站設(shè)計、ChatGPT

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)