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

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

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

<dependency>

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

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

 <version> 1.2.0 </version>

</dependency>

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

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

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

<Java代碼>

DataStream <String> input = ...;

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

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

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

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

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

最后一個(gè)配置選項(xiàng)是批量大小。這指定何時(shí)應(yīng)該關(guān)閉零件文件并啟動(dòng)一個(gè)新的零件。 （默認(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)建一個(gè)寫(xiě)入到遵循此模式的桶文件的接收器：

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

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

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

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

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

廣告

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