Flink slotSharingGroup disableChain startNewChain 用法案例

ゝ一纸荒年。 2023-06-11 03:17 89阅读 0赞

使用slot分组 或者 disableChaining()，我们可以将算子的依赖链给隔离或者分开，这样可以针对不同的需求进行不同的优化。

>     package application;
>     
>     
>     import com.alibaba.fastjson.JSONObject;
>     import operator.*;
>     import org.apache.commons.lang.StringUtils;
>     import org.apache.flink.api.common.functions.FilterFunction;
>     import org.apache.flink.api.common.functions.MapFunction;
>     import org.apache.flink.api.common.serialization.SimpleStringSchema;
>     import org.apache.flink.api.java.utils.ParameterTool;
>     import org.apache.flink.streaming.api.TimeCharacteristic;
>     import org.apache.flink.streaming.api.datastream.DataStream;
>     import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
>     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
>     import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
>     import org.apache.flink.streaming.api.windowing.time.Time;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
>     import org.apache.flink.util.OutputTag;
>     import org.slf4j.LoggerFactory;
>     
>     import java.util.Properties;
>     
>     
>     /**
>      * todo 备忘记录：使用分流方式，分为单场景快速流，分为多场景或者需要查询hbase的复合流
>      *
>      *  todo 1，接下来要做的；了解并行度；添加水印；添加checkpoint；代码正规化；
>      *
>      *  todo 2, 测试slot group ，通过 slotSharingGroup
>      *
>      *  todo 3, disableChain,startNewChain的使用
>      *
>      *  todo 4, name()可以取名字哦
>      */
>     public class StormToFlink_hbase_demo {
>     
>         private static org.slf4j.Logger logger = LoggerFactory.getLogger(StormToFlink_hbase_demo.class);
>     
>         public static void main(String[] args) throws Exception {
>     //        String fileUrl = "D:\\wxgz-local\\resources_yace\\";
>             String fileUrl = "/zywa/job/storm/resources_new/";
>     
>     
>             // todo 2，读取kafka数据
>             final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>             env.setParallelism(4);
>             //todo 获取kafka的配置属性
>             args = new String[]{"--input-topic", "topn_test", "--bootstrap.servers", "node2.hadoop:9092,node3.hadoop:9092",
>                     "--zookeeper.connect", "node1.hadoop:2181,node2.hadoop:2181,node3.hadoop:2181", "--group.id", "cc2"};
>     
>     
>             ParameterTool parameterTool = ParameterTool.fromArgs(args);
>             env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>             Properties sendPros = parameterTool.getProperties();
>             Properties pros = parameterTool.getProperties();
>     
>             //todo 指定输入数据为kafka topic
>             DataStream<String> kafkaDstream = env.addSource(new FlinkKafkaConsumer010<String>(
>                     pros.getProperty("input-topic"),
>                     new SimpleStringSchema(),
>                     pros).setStartFromLatest()
>     
>             ).setParallelism(4);
>     
>             //todo 定义一个测流输出
>             final OutputTag<JSONObject> mutiOutputTag = new OutputTag<JSONObject>("mutiStream") {
>             };
>     
>     
>     
>             //todo 2，过滤掉不满足格式的数据
>     //        DataStream<JSONObject> jsonDstream = kafkaDstream.map(new MapOperator_01(fileUrl)).disableChaining().setParallelism(4);
>             DataStream<JSONObject> jsonDstream = kafkaDstream.map(new MapOperator_01(fileUrl)).disableChaining().name("MapOperator_01").setParallelism(4);
>     
>     
>     //        SingleOutputStreamOperator<JSONObject> splitStream = jsonDstream.process(new ProcessOperator_01(mutiOutputTag)).startNewChain().setParallelism(4);
>             SingleOutputStreamOperator<JSONObject> splitStream = jsonDstream.process(new ProcessOperator_01(mutiOutputTag)).disableChaining().setParallelism(4);
>     
>             //todo 3,需要查询hbase的流
>             DataStream<JSONObject> mutiStream = splitStream.getSideOutput(mutiOutputTag);
>             mutiStream.print();
>     
>             //todo 4，先做单条件流 ,去匹配场景表达式
>             DataStream<JSONObject> filterDstream = splitStream.filter(new FilterOperator_01()).disableChaining().setParallelism(4);
>     //        DataStream<JSONObject> filterDstream = splitStream.filter(new FilterOperator_01()).slotSharingGroup("group_03").setParallelism(4);
>     
>             DataStream<JSONObject> mapDstream = filterDstream.map(new MapOperator_02()).name("MapOperator_02").setParallelism(4);
>     
>     
>             //todo 推送下发
>             mapDstream.filter(new FilterFunction<JSONObject>() {
>                 @Override
>                 public boolean filter(JSONObject json) throws Exception {
>                     //推送
>                     if (json.containsKey("Payload")) {
>                         return true;
>                     }
>                     return false;
>                 }
>             }).setParallelism(4)
>                     .map(new MapFunction<JSONObject, String>() {
>                         @Override
>                         public String map(JSONObject s) throws Exception {
>                             return s.toJSONString();
>                         }
>                     }).setParallelism(4)
>                     .addSink(new FlinkKafkaProducer010<String>(
>                             "dianyou_wx_test3",
>                             new SimpleStringSchema(),
>                             sendPros)).setParallelism(4);
>     
>             //todo 下发到kafka filter
>             SingleOutputStreamOperator processDstream = mapDstream.filter(new FilterFunction<JSONObject>() {
>                 @Override
>                 public boolean filter(JSONObject json) throws Exception {
>                     //推送
>                     if (!json.containsKey("Payload")) {
>                         return true;
>                     }
>                     return false;
>                 }
>             }).setParallelism(4)
>                     .keyBy(value -> value.getString("appKey"))
>     
>                     .window(TumblingProcessingTimeWindows.of(Time.milliseconds(500)))
>     
>                     .process(new ProcessOperator_02())
>                     .setParallelism(4);
>     
>     
>             //todo 发送到kafka
>             processDstream.addSink(new FlinkKafkaProducer010<String>(
>                     "dianyou_wx_test2",
>                     new SimpleStringSchema(),
>                     sendPros))
>                     .setParallelism(4);
>     
>     
>             //todo 匹配复杂情况
>     
>             DataStream<JSONObject> mutiProcessDstream = mutiStream.keyBy(value -> value.getString("appKey"))
>                     .window(TumblingProcessingTimeWindows.of(Time.milliseconds(500)))
>                     .process(new ProcessOperator_03())
>                     .setParallelism(4);
>     
>     
>             //todo 批量条件
>             DataStream<JSONObject> process = mutiProcessDstream.map(new MapOperator_03())
>                     .setParallelism(4)
>                     .filter(new FilterFunction<JSONObject>() {
>                         @Override
>                         public boolean filter(JSONObject jsonObject) throws Exception {
>                             if (jsonObject.containsKey("tiaojian")) {
>                                 return true;
>                             }
>                             return false;
>                         }
>                     }).setParallelism(4)
>     
>                     .keyBy(value -> value.getJSONObject("logJson").getString("appKey"))
>                     .window(TumblingProcessingTimeWindows.of(Time.milliseconds(500)))
>                     .process(new ProcessOperator_04())
>                     .setParallelism(4);
>     
>     
>             //todo 已经匹配到场景的情况下，先发送到topic
>     
>             process.map(new MapOperator_04())
>                     .setParallelism(4).
>                     filter(new FilterFunction<String>() {
>                         @Override
>                         public boolean filter(String value) throws Exception {
>                             if (StringUtils.isNotBlank(value)) {
>                                 return true;
>                             }
>                             return false;
>                         }
>                     })
>                     .setParallelism(4)
>                     .addSink(new FlinkKafkaProducer010<String>(
>                             "dianyou_wx_test3",
>                             new SimpleStringSchema(),
>                             sendPros)).setParallelism(4);
>     
>     
>             process.map(new MapOperator_05())
>                     .setParallelism(4).filter(new FilterFunction<String>() {
>                 @Override
>                 public boolean filter(String value) throws Exception {
>                     if (StringUtils.isNotBlank(value)) {
>                         return true;
>                     }
>                     return false;
>                 }
>             }).setParallelism(4)
>     
>                     .addSink(new FlinkKafkaProducer010<String>(
>                             "dianyou_wx_test2",
>                             new SimpleStringSchema(),
>                             sendPros)).setParallelism(4);
>     
>             env.execute("startExecute");
>         }
>     
>     
>     }

web端效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20230531/94ade16ee86a473db3821fb6255efda5.png