本文将分享一个Flink多维实时分析项目，也是最近正在做的，觉得比较有价值，记录一下，防止以后再次踩坑。主要涉及到的知识有 EventTime 和 ProcessTime 的使用，WaterMark 的使用，大数据量（一秒300MB）下维表 Join 的方案，定时器的使用，Flink DataStream 和 Flink Sql 的使用，Flink 任务调优等，干货多多！！！
一、需求背景
1. 数据来源
② 维度数据来自于 Hive
2. 数据格式
① kafka 数据
| imsi 移动用户识别码 |
ul_data 上行流量 |
dl_data 下行流量 |
cell_id 小区id |
|---|---|---|---|
| B001 | 100 | 230 | 112233 |
| B002 | 200 | 330 | 221134 |
② 小区维表
| cell_id 小区id |
scene_name 场景名称 |
其他字段 |
|---|---|---|
| 112233 | 4A级景区 | … |
| 221134 | 5A级景区 | … |
3. 输出结果
| date_time 窗口结束时间 |
scene_name 场景名称 |
user_cnt 用户数 |
flow_data 流量 |
|---|---|---|---|
| 202401311230 | 4A级景区 | 230213 | 121312233213 |
| 202401311235 | 5A级景区 | 330454 | 221323241334 |
4. 需求解读
需要注意的一点是维表会不定期更新，并不是一成不变的。
二、实现过程
1. 使用异步 Join
1public static void main(String[] args) throws Exception {
2 // 设置时间语义为processTime
3 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
4 FlinkKafkaConsumer<String> fkc = KafkaUtil.getKafkaSource(source_topic, "source_group");
5 // 读取kafka数据流
6 SingleOutputStreamOperator<String> sourceDs = env.addSource(fkc).name(source_topic + "_source");
7 // 解析kafka数据
8 SingleOutputStreamOperator<SourceDpiHttp> mapDs = sourceDs.flatMap(new MultipleMessageParser(net_type)).name("multiple_message_parser");
9 // 关联小区场景维度表
10 SingleOutputStreamOperator<SourceDpiHttp> cellDimDs = AsyncDataStream.unorderedWait(mapDs, new DimAsyncFunction<SourceDpiHttp>(DataSourceType.PRESTO.name(), DIM_GZDP_XUNQI_CELL) {
11
12 @SuppressWarnings("unchecked")
13 @Override
14 public Tuple2<String, String>[] getJoinCondition(SourceDpiHttp obj) {
15 Tuple2<String, String> tuple2 = Tuple2.of("id", String.valueOf(obj.getCellId()));
16 return new Tuple2[]{tuple2};
17 }
18
19 @Override
20 public void join(SourceDpiHttp model, JSONObject dimInfo) {
21 model.setSceneName(dimInfo.getString("scene_name"));
22 model.setCellName(dimInfo.getString("cell_name"));
23 model.setCellType(dimInfo.getString("cell_type"));
24 }
25 }, 60, TimeUnit.SECONDS).name("cell_dim_join");
26 // 设置waterMark时间字段和窗口大小
27 // 此处使用当前时间作为eventTime
28 SingleOutputStreamOperator<SourceDpiHttp> wmStream = cellDimDs
29 .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<SourceDpiHttp>() {
30 private Long currentMaxTimestamp = 0L;
31
32 @Override
33 public Watermark getCurrentWatermark() {
34 // 最大允许的消息延迟是0秒
35 Long maxOutOfOrderNess = 0L;
36 return new Watermark(currentMaxTimestamp - maxOutOfOrderNess);
37 }
38
39 @Override
40 public long extractTimestamp(SourceDpiHttp item, long l) {
41 long timestamp = 0;
42 currentMaxTimestamp = Math.max(item.getEndTime(), currentMaxTimestamp);
43 return timestamp;
44 }
45 }).name("window_setting");
46 // 将流转换为table
47 Table table = tableEnv.fromDataStream(wmStream, "imsi,ulData,dlData,localCity,roamingType,cellId,cellName,sceneName,cellType,regionName,netType,rowtime.rowtime");
48
49 // 计算场景维度
50 String scene_sql = "select \n" +
51 " TUMBLE_START(rowtime, INTERVAL '" + window_size + "' SECOND) AS windowStart, \n" +
52 " netType, " +
53 " '' as dateTime,\n" +
54 " sceneName,\n" +
55 " count(distinct imsi) as userCnt,\n" +
56 " (sum(dlData)+sum(ulData)) as flowData\n" +
57 "from " + table + "\n" +
58 "GROUP BY \n" +
59 " TUMBLE(rowtime, INTERVAL '" + window_size + "' SECOND), \n" +
60 " netType,\n" +
61 " sceneName\n";
62 Table result_scene = tableEnv.sqlQuery(scene_sql);
63 // 类型转换映射
64 TypeInformation<CaSceneResult> sceneTypes = new TypeHint<CaSceneResult>() {
65 }.getTypeInfo();
66
67 // 将table转换为流并输出结果到kafka
68 SingleOutputStreamOperator<String> sceneMapDs = tableEnv
69 .toAppendStream(result_scene, sceneTypes)
70 .map(new SceneDateTimeParser()).name(sinkTopic_scene + "_transfer");
71 sceneMapDs.addSink(KafkaUtil.getKafkaSink(sinkTopic_scene)).name(sinkTopic_cell + "_sink");
72 env.execute("main_cell_job");
73}
2. 使用广播流
3. 使用外挂程序
4. 使用 DataStream
1public static void main(String[] args) throws Exception {
2 // 设置时间语义为eventTime
3 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
4
5 FlinkKafkaConsumer<String> fkc = KafkaUtil.getKafkaSource(source_topic, consumerGroup);
6 // 读取kafka数据流
7 SingleOutputStreamOperator<String> sourceDs = env.addSource(fkc).name(source_topic + "_source").setParallelism(parallelism);
8 // 解析kafka数据
9 SingleOutputStreamOperator<SourceDpiHttp> mapDs = sourceDs.flatMap(new MultipleMessageParser()).setParallelism(parallelism).name("multiple_message_parser");
10 // 设置waterMark时间字段和窗口大小
11 SingleOutputStreamOperator<SourceDpiHttp> wmStream = mapDs
12 .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<SourceDpiHttp>() {
13 private Long currentMaxTimestamp = 0L;
14
15 @Override
16 public Watermark getCurrentWatermark() {
17 // 最大允许的消息延迟是0秒
18 Long maxOutOfOrderNess = waterMark * 1000;
19 return new Watermark(currentMaxTimestamp - maxOutOfOrderNess);
20 }
21
22 @Override
23 public long extractTimestamp(SourceDpiHttp item, long l) {
24 long timestamp = item.getEndTime();
25 currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
26 return timestamp;
27 }
28 }).name("water_mark");
29
30 // 计算场景维度
31 SingleOutputStreamOperator<String> sceneResultStream = wmStream.
32 .keyBy((KeySelector<SourceDpiHttp, String>) SourceDpiHttp::getSceneName)
33 .window(TumblingEventTimeWindows.of(Time.minutes(5)))
34 .process(new ProcessWindowFunction<SourceDpiHttp, String, String, TimeWindow>() {
35 @Override
36 public void process(String sceneKey, ProcessWindowFunction<SourceDpiHttp, String, String, TimeWindow>.Context context, Iterable<SourceDpiHttp> iterable, Collector<String> collector) throws Exception {
37 try {
38 long flowData = 0;
39 Set<String> uvSet = new HashSet<>();
40 long windowEnd = context.window().getEnd();
41 for (SourceDpiHttp sourceDpiHttp : iterable) {
42 flowData += (sourceDpiHttp.getDlData() + sourceDpiHttp.getUlData());
43 uvSet.add(sourceDpiHttp.getImsi());
44 }
45 CaSceneResult result = new CaSceneResult();
46 result.setDateTime(new SimpleDateFormat("yyyyMMddHHmm").format(windowEnd));
47 result.setNetType(netType);
48 result.setSceneName(sceneKey);
49 result.setUserCnt((long) uvSet.size());
50 result.setFlowData(flowData);
51 collector.collect(JSON.toJSONString(result));
52 } catch (Exception e) {
53 logger.error("场景维度计算出错：", e);
54 }
55 }
56 });
57
58 sceneResultStream.addSink(KafkaUtil.getKafkaSink(sinkTopic)).name(sinkTopic + "_sink").setParallelism(sinkParallelism);
59 env.execute("main_cell_job");
60}
5. 使用 ProcessTime + 定时器
1public class SceneKeyedProcessFunction extends KeyedProcessFunction<String, SourceDpiHttp, String> {
2 private static final Logger logger = LoggerFactory.getLogger(SceneKeyedProcessFunction.class);
3
4 private transient ValueState<Long> flowDataState;
5 private transient ValueState<List<String>> userCntState;
6 private transient ValueState<Boolean> timerRegisteredState;
7
8 private final String netType;
9 private final Integer windowSize;
10
11 public SceneKeyedProcessFunction(String netType, Integer windowSize) {
12 this.netType = netType;
13 this.windowSize = windowSize;
14 }
15
16 @Override
17 public void open(Configuration parameters) throws Exception {
18 flowDataState = getRuntimeContext().getState(new ValueStateDescriptor<>("flowData", Types.LONG));
19 userCntState = getRuntimeContext().getState(new ValueStateDescriptor<>("userCnt", Types.LIST(Types.STRING)));
20 timerRegisteredState = getRuntimeContext().getState(new ValueStateDescriptor<>("timerRegistered", Types.BOOLEAN));
21 }
22
23 @Override
24 public void processElement(SourceDpiHttp sourceDpiHttp, KeyedProcessFunction<String, SourceDpiHttp, String>.Context ctx, Collector<String> out) throws Exception {
25 try {
26 Long flowData = flowDataState.value();
27 List<String> userCnt = userCntState.value();
28
29 // 如果窗口内数据的和为null，表示第一次处理，初始化为0
30 if (flowData == null) {
31 flowData = 0L;
32 }
33 if (ObjectUtils.isEmpty(userCnt)) {
34 userCnt = new ArrayList<>();
35 }
36 userCnt.add(sourceDpiHttp.getImsi());
37 userCntState.update(userCnt);
38
39 if (!"VOLTE".equalsIgnoreCase(netType)
40 && ObjectUtils.isNotEmpty(sourceDpiHttp.getDlData())
41 && ObjectUtils.isNotEmpty(sourceDpiHttp.getUlData())) {
42 // 更新窗口内数据的和
43 flowData += (sourceDpiHttp.getDlData() + sourceDpiHttp.getUlData());
44 flowDataState.update(flowData);
45 }
46 // 注册5分钟后的定时器
47 Boolean timerRegistered = timerRegisteredState.value();
48 if (timerRegistered == null || !timerRegistered) {
49 ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + windowSize * 60 * 1000);
50 timerRegisteredState.update(true);
51 }
52 } catch (Exception e) {
53 logger.error("小区维度计算出错：{}", e, e);
54 }
55 }
56
57 @Override
58 public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
59 String sceneKey = ctx.getCurrentKey();
60 Long flowData = flowDataState.value();
61 Set<String> userCnt = new HashSet<>();
62 if (ObjectUtils.isNotEmpty(userCntState) && ObjectUtils.isNotEmpty(userCntState.value())) {
63 userCnt = new HashSet<>(userCntState.value());
64 }
65
66 CaSceneResult result = new CaSceneResult();
67 result.setDateTime(new SimpleDateFormat("yyyyMMddHHmm").format(timestamp));
68 result.setNetType(netType);
69 result.setSceneName(sceneKey);
70 result.setUserCnt((long) userCnt.size());
71 result.setFlowData(flowData);
72
73 // 输出统计数据
74 out.collect(JSON.toJSONString(result));
75 // 清除当前状态
76 clearState();
77 ctx.timerService().deleteProcessingTimeTimer(timestamp);
78 timerRegisteredState.update(false);
79 }
80
81 @Override
82 public void close() throws Exception {
83 super.close();
84 }
85
86 public void clearState() {
87 if (Objects.nonNull(flowDataState)) {
88 flowDataState.clear();
89 }
90 if (Objects.nonNull(userCntState)) {
91 userCntState.clear();
92 }
93 }
94}
6. 使用 ProcessTime + 滑动窗口
这里直接使用 ProcessTime 对应的滑动窗口实现，大体代码如下：
1public static void main() throws Exception {
2 // 设置时间语义为processTime
3 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
4
5 FlinkKafkaConsumer<String> fkc = KafkaUtil.getKafkaSource(source_topic, consumerGroup);
6 // 读取kafka数据流
7 SingleOutputStreamOperator<String> sourceDs = env.addSource(fkc).name(source_topic + "_source").setParallelism(parallelism);
8 // 解析kafka数据
9 SingleOutputStreamOperator<SourceDpiHttp> mapDs = sourceDs.flatMap(new MultipleMessageParser()).setParallelism(parallelism).name("multiple_message_parser");
10
11 // 计算场景维度
12 SingleOutputStreamOperator<String> sceneResultStream = mapDs.filter((FilterFunction<SourceDpiHttp>) item -> ObjectUtils.isNotEmpty(item.getSceneName()) && !"000000".equals(item.getSceneName()))
13 .keyBy((KeySelector<SourceDpiHttp, String>) SourceDpiHttp::getSceneName)
14 .window(TumblingProcessingTimeWindows.of(Time.minutes(windowSize)))
15 .apply(new WindowFunction<SourceDpiHttp, String, String, TimeWindow>() {
16 @Override
17 public void apply(String sceneKey, TimeWindow window, Iterable<SourceDpiHttp> input, Collector<String> out) throws Exception {
18 try {
19 long flowData = 0;
20 Set<String> uvSet = new HashSet<>();
21 long windowEnd = window.getEnd();
22 for (SourceDpiHttp sourceDpiHttp : input) {
23 flowData += (sourceDpiHttp.getDlData() + sourceDpiHttp.getUlData());
24 uvSet.add(sourceDpiHttp.getImsi());
25 }
26 CaSceneResult result = new CaSceneResult();
27 result.setDateTime(new SimpleDateFormat("yyyyMMddHHmm").format(windowEnd));
28 result.setNetType(netType);
29 result.setSceneName(sceneKey);
30 result.setUserCnt((long) uvSet.size());
31 result.setFlowData(flowData);
32 out.collect(JSON.toJSONString(result));
33 } catch (Exception e) {
34 logger.error("场景维度计算出错：", e);
35 }
36 }
37 });
38
39 sceneResultStream.addSink(KafkaUtil.getKafkaSink(sink_kafka_server, sinkTopic_scene)).name(sinkTopic_scene + "_sink").setParallelism(sinkParallelism);
40 env.execute("main_cell_job");
41}
三、调优部分
1. checkpoint 调优
2. 设置合适的重试策略
3. 调整并行度
4. 调整 task manager 的内存
5. 调整 WaterMark
6. 调整 kafka 参数
四、写在最后
往期推荐
原文作者：毛毛小妖的笔记；如有侵权，请联系管理员删除。
下一篇：
5种常见的Flink维表Join方案
请登录后发表评论
登录
全部评论
