Flink slotSharingGroup disableChain startNewChain 用法案例
使用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端效果：