歡迎訪問(wèn)我的GitHubhttps://github.com/zq2599/blog_demos 內(nèi)容：所有原創(chuàng)文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等； 歡迎訪問(wèn)我的GitHub
Flink處理函數(shù)實(shí)戰(zhàn)系列鏈接本篇概覽
編碼實(shí)戰(zhàn)接下來(lái)咱們開發(fā)一個(gè)應(yīng)用來(lái)體驗(yàn)CoProcessFunction，功能非常簡(jiǎn)單，描述如下：
源碼下載如果您不想寫代碼，整個(gè)系列的源碼可在GitHub下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：
這個(gè)git項(xiàng)目中有多個(gè)文件夾，本章的應(yīng)用在flinkstudy文件夾下，如下圖紅框所示： Map算子
package com.bolingcavalry.coprocessfunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.StringUtils;
public class WordCountMap implements MapFunction<String, Tuple2<String, Integer>> {
@Override
public Tuple2<String, Integer> map(String s) throws Exception {
if(StringUtils.isNullOrWhitespaceOnly(s)) {
System.out.println("invalid line");
return null;
}
String[] array = s.split(",");
if(null==array || array.length<2) {
System.out.println("invalid line for array");
return null;
}
return new Tuple2<>(array[0], Integer.valueOf(array[1]));
}
}便于擴(kuò)展的抽象類
package com.bolingcavalry.coprocessfunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
/**
* @author will
* @email zq2599@gmail.com
* @date 2020-11-09 17:33
* @description 串起整個(gè)邏輯的執(zhí)行類，用于體驗(yàn)CoProcessFunction
*/
public abstract class AbstractCoProcessFunctionExecutor {
/**
* 返回CoProcessFunction的實(shí)例，這個(gè)方法留給子類實(shí)現(xiàn)
* @return
*/
protected abstract CoProcessFunction<
Tuple2<String, Integer>,
Tuple2<String, Integer>,
Tuple2<String, Integer>> getCoProcessFunctionInstance();
/**
* 監(jiān)聽根據(jù)指定的端口，
* 得到的數(shù)據(jù)先通過(guò)map轉(zhuǎn)為Tuple2實(shí)例，
* 給元素加入時(shí)間戳，
* 再按f0字段分區(qū)，
* 將分區(qū)后的KeyedStream返回
* @param port
* @return
*/
protected KeyedStream<Tuple2<String, Integer>, Tuple> buildStreamFromSocket(StreamExecutionEnvironment env, int port) {
return env
// 監(jiān)聽端口
.socketTextStream("localhost", port)
// 得到的字符串"aaa,3"轉(zhuǎn)成Tuple2實(shí)例，f0="aaa"，f1=3
.map(new WordCountMap())
// 將單詞作為key分區(qū)
.keyBy(0);
}
/**
* 如果子類有側(cè)輸出需要處理，請(qǐng)重寫此方法，會(huì)在主流程執(zhí)行完畢后被調(diào)用
*/
protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) {
}
/**
* 執(zhí)行業(yè)務(wù)的方法
* @throws Exception
*/
public void execute() throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 并行度1
env.setParallelism(1);
// 監(jiān)聽9998端口的輸入
KeyedStream<Tuple2<String, Integer>, Tuple> stream1 = buildStreamFromSocket(env, 9998);
// 監(jiān)聽9999端口的輸入
KeyedStream<Tuple2<String, Integer>, Tuple> stream2 = buildStreamFromSocket(env, 9999);
SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream = stream1
// 兩個(gè)流連接
.connect(stream2)
// 執(zhí)行低階處理函數(shù)，具體處理邏輯在子類中實(shí)現(xiàn)
.process(getCoProcessFunctionInstance());
// 將低階處理函數(shù)輸出的元素全部打印出來(lái)
mainDataStream.print();
// 側(cè)輸出相關(guān)邏輯，子類有側(cè)輸出需求時(shí)重寫此方法
doSideOutput(mainDataStream);
// 執(zhí)行
env.execute("ProcessFunction demo : CoProcessFunction");
}
}
子類決定CoProcessFunction的功能
package com.bolingcavalry.coprocessfunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CollectEveryOne extends AbstractCoProcessFunctionExecutor {
private static final Logger logger = LoggerFactory.getLogger(CollectEveryOne.class);
@Override
protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() {
return new CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() {
@Override
public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {
logger.info("處理1號(hào)流的元素：{},", value);
out.collect(value);
}
@Override
public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {
logger.info("處理2號(hào)流的元素：{}", value);
out.collect(value);
}
};
}
public static void main(String[] args) throws Exception {
new CollectEveryOne().execute();
}
}
驗(yàn)證
12:45:38,774 INFO CollectEveryOne - 處理1號(hào)流的元素：(aaa,111), (aaa,111) 12:45:43,816 INFO CollectEveryOne - 處理2號(hào)流的元素：(bbb,222) (bbb,222) 更多
|
|
|
