Merge remote-tracking branch 'origin/master'

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

@@ -49,7 +49,7 @@ public class LaneNormJob {
             KeySelector<TravelEvent, String> carKeySelector = new KeySelector<TravelEvent, String>() {
                 @Override
                 public String getKey(TravelEvent car) throws Exception {
-                    return car.toString();
+                    return car.getCrossId();
                 }
             };
             MapFunction<String, Cycle> getCycleFunction = new MapFunction<String, Cycle>() {
@@ -82,8 +82,8 @@ public class LaneNormJob {
             WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car, timestamp) -> Long.parseLong(car.getEventTime()));
             WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> Long.parseLong(cycle.getEndDateTime()));
             //根据corssid 分组 保证同一路口的数据进行一个进程
-            DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy).keyBy(cycleKeySelect);
-            DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).keyBy(carKeySelector).filter(filterFunction);
+            DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy);
+            DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).filter(filterFunction);
             ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
             DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties",args, "app_cross_dimension_norm");
             cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4)

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java

@@ -6,6 +6,7 @@ import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
+import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -188,6 +189,8 @@ public class FlinkUtils {
         env.setStateBackend(new FsStateBackend(checkpointPath));
         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
         env.setRestartStrategy(RestartStrategies.noRestart());
+        //修改Checkpoint的模式 时廷和吞吐量要求非常高但对数据的一致性要求不高的场景。flink内部无须exactly-once检查
+        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
         // 基于Kafka 原始JSON格式的字符串，获取消息流
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
         kafkaConsumer.setStartFromLatest();

realtime/hologram-streaming/src/main/resources/doris.properties

 driver=com.mysql.jdbc.Driver
-host=13.145.183.136
+host=10.243.0.22
 query.port=9030
 http.port=8030
 username=root
-password=Zhht2023!
+password=mima
 database=qxlk

realtime/hologram-streaming/src/main/resources/kafka.properties

@@ -2,6 +2,6 @@ bootstrap.servers=139.9.157.176:9092
 kafka.input.topics=trips_info
 auto.offset.reset=latest
 group.id=TravelInfoJob
-checkpoint.path=file:///data/flink/state
+checkpoint.path=hdfs://10.243.0.24:9000/data/flink/state
 checkpoint.interval=60000
 enable.auto.commit=true