Flink中怎么使用split

這篇文章主要講解了“Flink中怎么使用split”，文中的講解內(nèi)容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Flink中怎么使用split”吧！

網(wǎng)站建設(shè)哪家好，找創(chuàng)新互聯(lián)！專注于網(wǎng)頁設(shè)計、網(wǎng)站建設(shè)、微信開發(fā)、小程序定制開發(fā)、集團企業(yè)網(wǎng)站建設(shè)等服務項目。為回饋新老客戶創(chuàng)新互聯(lián)還提供了昌邑免費建站歡迎大家使用！

flink的神奇分流器-sideoutput

這個可以用來分流，很方便的一次就可以對數(shù)據(jù)進行篩選返回。

還有針對算法處理的迭代操作，我們已經(jīng)講過兩篇文章了：

Flink特異的迭代操作-bulkIteration

不得不會的Flink Dataset的DeltaI迭代操作

一個是全量迭代，一個是增量迭代。

還有優(yōu)秀又雞肋的watermark機制

不懂watermark？來吧～

對于迭代操作，其實還有一講，那就是流處理的迭代操作。那么本文就針對這個進行分析～

Flink的迭代流程序?qū)嶋H就是實現(xiàn)了一個步進函數(shù)，然后將其嵌入到IterativeStream內(nèi)部。要知道Flink的Datastream正常情況下是不會結(jié)束的，所以也沒有所謂的最大迭代次數(shù)。這種情況下，你需要自己指定哪個類型的數(shù)據(jù)需要回流去繼續(xù)迭代，哪個類型的數(shù)據(jù)繼續(xù)向下傳輸，這個分流的方式有兩種：split和filter，官方網(wǎng)站在介紹迭代流的時候使用的是filter。我們這里就先按照官網(wǎng)的介紹走，然后案例展示的時候使用split給大家做個demo。

首先，要創(chuàng)建一個IterativeStream

IterativeStream<Integer> iteration =input.iterate();

接著就可以定義對該留要進行的邏輯操作，官網(wǎng)這里就很簡單的舉了一個map的例子。

DataStream<Integer> iterationBody =iteration.map(/* this is executed many times */);

調(diào)用IterativeStream的closeWith(feedbackStream)方法可以對迭代流進行閉環(huán)操作。傳遞給closeWith函數(shù)的DataStream會返回值迭代的頭部。常用的做法是用filter來分離流的向后迭代的部分和向前傳遞的部分。。

iteration.closeWith(iterationBody.filter(/*one part of the stream */));

DataStream<Integer> output =iterationBody.filter(/* some other part of the stream */);

官方給了一個連續(xù)不斷減1直到數(shù)據(jù)為零的例子：

DataStream<Long> someIntegers =env.generateSequence(0, 1000);

// 創(chuàng)建迭代流

IterativeStream<Long> iteration =someIntegers.iterate();

// 增加處理邏輯，對元素執(zhí)行減一操作。

DataStream<Long> minusOne =iteration.map(new MapFunction<Long, Long>() {

@Override

public Long map(Long value) throws Exception {

  return value - 1 ;

 }

});

// 獲取要進行迭代的流，

DataStream<Long> stillGreaterThanZero= minusOne.filter(new FilterFunction<Long>() {

@Override

public boolean filter(Long value) throws Exception {

  return (value > 0);

 }

});

// 對需要迭代的流形成一個閉環(huán)

iteration.closeWith(stillGreaterThanZero);

// 小于等于0的數(shù)據(jù)繼續(xù)向前傳輸

DataStream<Long> lessThanZero =minusOne.filter(new FilterFunction<Long>() {

@Override

public boolean filter(Long value) throws Exception {

  return (value <= 0);

 }

});

感謝各位的閱讀，以上就是“Flink中怎么使用split”的內(nèi)容了，經(jīng)過本文的學習后，相信大家對Flink中怎么使用split這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！

當前名稱：Flink中怎么使用split
分享鏈接：http://bm7419.com/article28/pccecp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供全網(wǎng)營銷推廣、移動網(wǎng)站建設(shè)、自適應網(wǎng)站、、網(wǎng)站維護、外貿(mào)建站

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

猜你還喜歡下面的內(nèi)容