优化按方向计算的任务

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/DirectionEvalJob.java

@@ -37,39 +37,24 @@ public class DirectionEvalJob {
         // 获取信控数据-周期结束实时流 Kafka
         DataStream<Cycle> cycleStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
-                        .map(record -> {
-                            return JSON.parseObject(record, Cycle.class);
-                        });
+                        .map(record -> JSON.parseObject(record, Cycle.class));
 
         // 获取旅行事件数据实时流 kafka
         DataStream<TravelEvent> travelEventStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
-                        .map(record -> {
-                            return JSON.parseObject(record, TravelEvent.class);
-                        }).filter(event -> {
-                            return StringUtils.isNotEmpty(event.getEventLineId());
-                        });
+                        .map(record -> JSON.parseObject(record, TravelEvent.class)).filter(event -> StringUtils.isNotEmpty(event.getEventLineId()));
 
-        WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy
-                .<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3)))
-                .withTimestampAssigner((cycle, timestamp) -> Long.parseLong(cycle.getEndDateTime()))
-                .withIdleness(Duration.ofSeconds(600));
-
-        WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy
-                .<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
-                .withTimestampAssigner((xx, timestamp) -> Long.parseLong(xx.getEventTime()));
 
         // 周期数据 interval join 事件流数据，计算 交通流量和排队长度两个指标
         // 旅行事件数据流
-        KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream.assignTimestampsAndWatermarks(travelEventWatermarkStrategy)
+        KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream
                 .keyBy(record -> record.getCrossId());
 
         // 定义好doris sink builder
         DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_direction_eval_second_metric");
 
         // 周期数据join 旅行事件
-        cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
-                .keyBy(record -> record.getCrossCode())
+        cycleStream.keyBy(record -> record.getCrossCode())
                 .connect(travelEventKeyedStream)
                 .process(new CalDirectionSecondMetricFunction())
                 .map((MapFunction<DirectionEvalSecondMetric, String>) desm -> {