您好,登錄后才能下訂單哦!
這篇文章主要介紹“Flink Process怎么用”,在日常操作中,相信很多人在Flink Process怎么用問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”Flink Process怎么用”的疑惑有所幫助!接下來,請跟著小編一起來學習吧!
process算子:處理每個keyBy(分區(qū))輸入到窗口的批量數(shù)據(jù)流(為KeyedStream類型數(shù)據(jù)流)
示例環(huán)境
java.version: 1.8.x flink.version: 1.11.1
示例數(shù)據(jù)源 (項目碼云下載)
Flink 系例 之 搭建開發(fā)環(huán)境與數(shù)據(jù)
Process.java
import com.flink.examples.DataSource; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.windows.GlobalWindow; import org.apache.flink.util.Collector; import java.util.Iterator; import java.util.List; /** * @Description process算子:處理每個keyBy(分區(qū))輸入到窗口的批量數(shù)據(jù)流(為KeyedStream類型數(shù)據(jù)流) */ public class Process { /** * 遍歷集合,分別打印不同性別的總?cè)藬?shù)與年齡之和 * @param args * @throws Exception */ public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); List<Tuple3<String, String, Integer>> tuple3List = DataSource.getTuple3ToList(); DataStream<String> dataStream = env.fromCollection(tuple3List) .keyBy((KeySelector<Tuple3<String, String, Integer>, String>) k -> k.f1) //按數(shù)量窗口滾動,每3個輸入數(shù)據(jù)流,計算一次 .countWindow(3) //處理每keyBy后的窗口數(shù)據(jù)流,process方法通常應用于KeyedStream類型的數(shù)據(jù)流處理 .process(new ProcessWindowFunction<Tuple3<String, String, Integer>, String, String, GlobalWindow>() { /** * 處理窗口數(shù)據(jù)集合 * @param s 從keyBy里返回的key值 * @param context 窗口的上下文 * @param input 從窗口獲取的所有分區(qū)數(shù)據(jù)流 * @param out 輸出數(shù)據(jù)流對象 * @throws Exception */ @Override public void process(String s, Context context, Iterable<Tuple3<String, String, Integer>> input, Collector<String> out) throws Exception { Iterator<Tuple3<String, String, Integer>> iterator = input.iterator(); int total = 0; int i = 0; while (iterator.hasNext()){ Tuple3<String, String, Integer> tuple3 = iterator.next(); total += tuple3.f2; i ++ ; } out.collect(s + "共:"+i+"人,平均年齡:" + total/i); } }); dataStream.print(); env.execute("flink Process job"); } }
打印結(jié)果
4> girl共:3人,平均年齡:24 2> man共:3人,平均年齡:26
到此,關(guān)于“Flink Process怎么用”的學習就結(jié)束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習,快去試試吧!若想繼續(xù)學習更多相關(guān)知識,請繼續(xù)關(guān)注億速云網(wǎng)站,小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。