Merge remote-tracking branch 'origin/master'

realtime/hologram-streaming/pom.xml

@@ -60,6 +60,12 @@
         </dependency>
 
         <dependency>
+            <groupId>org.apache.doris</groupId>
+            <artifactId>flink-doris-connector-1.14_${scala.binary.version}</artifactId>
+            <version>1.1.0</version>
+        </dependency>
+
+        <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
             <version>1.2.32</version>

realtime/hologram-streaming/src/main/java/com/zhht/irn/consts/Constant.java

@@ -12,9 +12,4 @@ public class Constant {
     public static final String TRAVEL_TOPIC_NAME = "trips_info";
 
     public static final String TRAVEL_EVENT_TOPIC_NAME = "trips_event_info";
-
-    // 周期计算的延迟时间 15秒
-    public static final int CYCLE_CALCULATE_LATENESS = 15 * 1000;
-
-    public static final int INSERT_BATCH_SIZE = 100;
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Travel.java

@@ -116,8 +116,8 @@ public class Travel {
     //实际停车次数
     private int stopCount;
 
-    //停车总时长
-    private double stopTime;
+    //停车总时长, 单位毫秒
+    private int stopTime;
 
     //平均车速
     private double avgSpeed;
@@ -392,11 +392,11 @@ public class Travel {
         this.stopCount = stopCount;
     }
 
-    public double getStopTime() {
+    public int getStopTime() {
         return stopTime;
     }
 
-    public void setStopTime(double stopTime) {
+    public void setStopTime(int stopTime) {
         this.stopTime = stopTime;
     }
 

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/Location.java

@@ -2,6 +2,8 @@ package com.zhht.irn.entity.dto;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
 
 /**
  * 位置信息
@@ -11,6 +13,7 @@ import lombok.Data;
  **/
 @Data
 @AllArgsConstructor
+@NoArgsConstructor
 public class Location implements Comparable<Location>{
 
    private Double longitude  ;//可选 经度

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelInfo.java

@@ -11,7 +11,7 @@ import java.util.List;
  * @create 2022-11-14 13:33{
  **/
 @Data
-public class TravelInfo {
+public class TravelInfo implements Comparable<TravelInfo>  {
 
     private Integer id  ;//必填 ID
     private String crossId  ;//必填 路口ID
@@ -31,6 +31,7 @@ public class TravelInfo {
     private Long arrivedTime  ;//可选 到达时间，毫秒级别时间戳
     private Double arrivedSpeed  ;//可选 到达时速度，单位km/h
     private String inCrossLineId  ;//可选 进入路口时车道编号
+    private String flowDirection  ;//流向
     private Long inCrossTime  ;//可选 进入路口时间，毫秒级别时间戳
     private Double inSpeed  ;//可选 进入时速度，单位km/h
     private String outCrossLineId  ;//可选 通过路口时车道编号
@@ -45,4 +46,9 @@ public class TravelInfo {
     private Double crossingTime  ;//可选 旅行时长，单位 秒 (s)（驶离时刻-到达时刻）
     private List<Location> locations  ;//必填 数组，参考 位置信息字段
     private String remark  ;//可选 备注
+
+    @Override
+    public int compareTo(TravelInfo o) {
+        return (int)(inCrossTime-o.getInCrossTime());
+    }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/metric/DirectionEvalSecondMetric.java

 package com.zhht.irn.entity.metric;
 
+import java.util.Date;
+
 /**
  * 按方向指标计算结果实体类
  *
@@ -116,4 +118,6 @@ public class DirectionEvalSecondMetric {
                 ", accStopCount=" + accStopCount +
                 '}';
     }
+
+
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalDirectionSecondMetricFunction.java

@@ -138,7 +138,6 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
         listState.clear();
         for(Map.Entry<String, Boolean> en : globalRecordStopStatusMap.entrySet()) {
             if(en.getValue() == true) {
-                System.out.println("stay stop status car record id:" + en.getKey());
                 listState.add(en.getKey());
             }
         }
@@ -146,7 +145,7 @@ public class CalDirectionSecondMetricFunction extends KeyedCoProcessFunction<Str
         if(cachedTravelEvent != null) {
             List<TravelEvent> newOne = new ArrayList<>();
             for(TravelEvent t : cachedTravelEvent) {
-                // 删除掉这个周期结束时间之间的数据
+                // 删除掉这个周期结束时间之前的数据
                 if(Long.parseLong(t.getEventTime()) >= Long.parseLong(cycle.getEndDateTime())) {
                     newOne.add(t);
                 }

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

@@ -6,23 +6,23 @@ import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
 import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
 import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
-import com.zhht.irn.sink.DirectionEvalSecondSink;
-import com.zhht.irn.source.CycleMockSource;
-import com.zhht.irn.source.TravelEventMockSource;
+import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.FlinkUtils;
 import com.zhht.irn.utils.StringUtils;
+import org.apache.doris.flink.sink.DorisSink;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.KeyedStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
-import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
 
+import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
+
 /**
  * 按方向评价实时任务，按秒计算
  *
@@ -43,7 +43,6 @@ public class DirectionEvalJob {
         DataStream<Cycle> cycleStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
                         .map(record -> {
-                    System.out.println(record);
                             return JSON.parseObject(record, Cycle.class);
                         });
 
@@ -51,7 +50,6 @@ public class DirectionEvalJob {
         DataStream<TravelEvent> travelEventStream =
                 FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
                         .map(record -> {
-                    System.out.println(record);
                             return JSON.parseObject(record, TravelEvent.class);
                         }).filter(event -> {
                             return StringUtils.isNotEmpty(event.getEventLineId());
@@ -71,12 +69,21 @@ public class DirectionEvalJob {
         KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream.assignTimestampsAndWatermarks(travelEventWatermarkStrategy)
                 .keyBy(record -> record.getCrossId());
 
+        // 定义好doris sink builder
+        DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-direction-eval.properties");
+
         // 周期数据join 旅行事件
         cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
                 .keyBy(record -> record.getCrossCode())
                 .connect(travelEventKeyedStream)
                 .process(new CalDirectionSecondMetricFunction())
-                .addSink(new DirectionEvalSecondSink()).setParallelism(4);
+                .map((MapFunction<DirectionEvalSecondMetric, String>) desm -> {
+                    // 将结果实体类，拼接成制表符分割的字符串，进行sink，要保持字段顺序跟表的字段顺序相一致
+                    return desm.getCrossId() + "\t" + desm.getLineId() + "\t" + desm.getCycleOrder() + "\t" + longToStandardFormatString(desm.getCycleStart()) + "\t"
+                            + longToStandardFormatString(desm.getCycleEnd()) + "\t" + desm.getCurrentSecond() + "\t" + desm.getAccArrivedFlow() + "\t"
+                            + desm.getAccQueueLength() + "\t" + desm.getStopFlagCount() + "\t" + desm.getAccStopCount() + "\t"
+                            + longToStandardFormatString(System.currentTimeMillis()) + "\t" + longToStandardFormatString(System.currentTimeMillis());
+                }).sinkTo(builder.build());
 
         // 旅行数据流
         env.execute("DirectionEvalJob");

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

@@ -11,6 +11,7 @@ import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.FilterFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.streaming.api.datastream.*;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -29,9 +30,11 @@ public class LaneNormJob {
     public static void main(String[] args) throws Exception {
         try {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-            env.setRestartStrategy(RestartStrategies.noRestart());
-            DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
-            DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
+            //env.setRestartStrategy(RestartStrategies.noRestart());
+            //设置重启策略
+            //env.setRestartStrategy( RestartStrategies.fixedDelayRestart(10, Time.seconds(10)));
+            DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"signal_cycle_data","laneNormJob1");
+            DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"trips_event_info","laneNormJob1");
             KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
                 @Override
                 public String getKey(Cycle cycle) throws Exception {
@@ -79,7 +82,7 @@ public class LaneNormJob {
 
             ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
             SingleOutputStreamOperator<List<LaneNorm>> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
-            process.addSink(new LaneNormSink()).name("laneNormJobSink");
+            process.addSink(new LaneNormSink()).name(" ");
             env.execute("laneNormJob");
         }catch (Exception e){
             System.out.println("程序异常关闭->"+e);

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

@@ -5,8 +5,10 @@ import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.sink.TravelMetricSink;
-import com.zhht.irn.utils.FileUtils;
+import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.FlinkUtils;
+import org.apache.doris.flink.sink.DorisSink;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -16,6 +18,7 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 
 import static com.zhht.irn.utils.StringUtils.isNotEmpty;
+import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
 
 /**
  * 旅行信息实时任务
@@ -34,12 +37,17 @@ public class TravelInfoJob {
         DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_TOPIC_NAME, "TravelInfoJob");
 
         // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查，需要时打开
-        // rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
+        // rawStringStream.sinkTo(FileUtils.getFileSink("./data/
 
-        // 开始计算旅行信息相关指标
-        rawStringStream.map((MapFunction<String, Travel>) s -> {
+        // 获取builder
+        DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-travel-info.properties");
+
+        DataStream<Travel> travelStream = rawStringStream
                 // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
-            Travel travel = JSON.parseObject(s, Travel.class);
+                .map(record -> JSON.parseObject(record, Travel.class))
+                .filter(travel -> travel.getLocations() != null);
+        // 开始计算旅行信息相关指标
+        travelStream.map(travel -> {
                 // -1 代表该值从未被处理过
             double minSpeed = -1, maxSpeed = -1, totalSpeed = -1;
 
@@ -85,7 +93,7 @@ public class TravelInfoJob {
             boolean stopped = false;
 
             List<Location> locations = travel.getLocations();
-            long stopTime = 0;
+            int stopTime = 0;
             int stopCount = 0;
 
             for(int j = 0; j < locations.size(); j ++) {
@@ -123,7 +131,19 @@ public class TravelInfoJob {
                     "停车总时长：{}", travel.getRecordId(), maxSpeed, minSpeed, travel.getAvgSpeed(), stopCount, stopTime);
 
             return travel;
-        }).addSink(new TravelMetricSink());
+        }).map((MapFunction<Travel, String>) travel -> {
+            return travel.getRecordId() + "\t" + travel.getCrossId() + "\t" + longToStandardFormatString(travel.getRecordTime()) + "\t"
+                    + longToStandardFormatString(travel.getTravelBeginTime()) + "\t" + longToStandardFormatString(travel.getTravelEndTime()) + "\t"
+                    + travel.getFlowDirection() + "\t" + travel.getCarId() + "\t" + travel.getCarColor() + "\t" + travel.getPlate() + "\t"
+                    + travel.getPlateColor() + "\t" + travel.getCarType() + "\t" + travel.getFirstLineId() + "\t" + longToStandardFormatString(travel.getFirstTime()) + "\t"
+                    + travel.getFirstSpeed() + "\t" + travel.getArrivedLineId() + "\t" + longToStandardFormatString(travel.getArrivedTime()) + "\t"
+                    + travel.getArrivedSpeed() + "\t" + travel.getInCrossLineId() + "\t" + longToStandardFormatString(travel.getInCrossTime()) + "\t"
+                    + travel.getInSpeed() + "\t" + travel.getOutCrossLineId() + "\t" + longToStandardFormatString(travel.getOutCrossTime()) + "\t"
+                    + travel.getOutSpeed() + "\t" + travel.getAwayLineId() + "\t" + longToStandardFormatString(travel.getAwayTime()) + "\t"
+                    + travel.getAwaySpeed() + "\t" + travel.getLastLineId() + "\t" + longToStandardFormatString(travel.getLastTime()) + "\t"
+                    + travel.getLastSpeed() + "\t" + travel.getCrossingTime() + "\t" + travel.getMinSpeed() + "\t" + travel.getMaxSpeed() + "\t"
+                    + travel.getAvgSpeed() + "\t" + travel.getStopCount() + "\t" + travel.getStopTime() + "\t" + longToStandardFormatString(System.currentTimeMillis());
+        }).sinkTo(builder.build());
 
         env.execute("TravelInfoJob");
     }

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

@@ -2,21 +2,33 @@ package com.zhht.irn.job;
 
 import com.zhht.irn.entity.dto.CycleSignalData;
 import com.zhht.irn.entity.dto.TravelEvent;
+import com.zhht.irn.entity.dto.TravelInfo;
+import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
 import com.zhht.irn.functions.TravelEventAndCycleCoProcessFunction;
 import com.zhht.irn.schema.CycleSignalKafkaSchema;
 import com.zhht.irn.schema.TravelEventKafkaSchema;
+import com.zhht.irn.schema.TravelInfoKafkaSchema;
+import com.zhht.irn.sink.TravelLaneInfoSink;
+import com.zhht.irn.utils.DorisUtils;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.functions.KeySelector;
+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.ConnectedStreams;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
+
+import java.io.InputStream;
+import java.util.List;
 import java.util.Properties;
 
 /**
@@ -26,85 +38,88 @@ import java.util.Properties;
  * @create 2022-11-14 11:43
  **/
 public class TravelSituationAnalysisJob {
-
     public static void main(String[] args)  {
 
         try {
 
+            InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("kafka.properties");
+            ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
+
+
 //1、创建环境
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             // checkpoint配置
             CheckpointConfig checkpointConfig = env.getCheckpointConfig();
             // checkpoint时间间隔3分钟
-           checkpointConfig.setCheckpointInterval(1* 60 * 1000);
+            checkpointConfig.setCheckpointInterval( 60 * 1000);
 
             // 两次checkpoint中最短时间间隔1分钟
             checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
             // 同时同时允许1个checkpoint进行
             checkpointConfig.setMaxConcurrentCheckpoints(1);
 
+            //这如果不指定使用FsStateBackend 则默认使用flink集群中的配置，rocksDB
+         //   env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
+            env.setStateBackend(new FsStateBackend(tool.getRequired("checkpoint.path")));
             env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+            env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
-            //设置重启策略
-            env.setRestartStrategy( RestartStrategies.fixedDelayRestart(5, Time.seconds(10)));
+            //设置重启策略 这里不设置重启策略，由平台统一拉起任务
+            // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
+            env.setRestartStrategy(RestartStrategies.noRestart());
 
-/*
-           关于flink checkpoint存储的设置统一在集群配置文件中设置
-            checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
-            env.setStateBackend(new RocksDBStateBackend("file:///opt/rocksDb/", true));*/
 
 
             Properties kafkaProperties = new Properties();
-         kafkaProperties.setProperty("bootstrap.servers", "dn1.zhht:9092,dn2.zhht:9092,dn3.zhht:9092");
-         //  kafkaProperties.setProperty("bootstrap.servers", "172.25.1.251:9092,172.25.1.122:9092,172.25.1.67:9092");
-            kafkaProperties.setProperty("group.id", "travel_event");
+           //  kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
+             kafkaProperties.setProperty("bootstrap.servers", tool.getRequired("bootstrap.servers"));
+          // kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
+            //  kafkaProperties.setProperty("group.id", "travel_event");
+            kafkaProperties.setProperty("group.id", "trips_info2");
             kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
             kafkaProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
 
 
             Properties kafkaProperties2 = new Properties();
-           kafkaProperties2.setProperty("bootstrap.servers", "dn1.zhht:9092,dn2.zhht:9092,dn3.zhht:9092");
-         //   kafkaProperties2.setProperty("bootstrap.servers", "172.25.1.251:9092,172.25.1.122:9092,172.25.1.67:9092");
-            kafkaProperties2.setProperty("group.id", "cycle_signal");
+          //  kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
+            kafkaProperties2.setProperty("bootstrap.servers", tool.getRequired("bootstrap.servers"));
+           // kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
+            kafkaProperties2.setProperty("group.id", "cycle_signal2");
             kafkaProperties2.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
             kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
 
-            //旅行事件
-            FlinkKafkaConsumerBase<TravelEvent> travelEvent =
-                    new FlinkKafkaConsumer<TravelEvent>("trips_event_info",
-                            new TravelEventKafkaSchema(),
-                            kafkaProperties) {
-                    };
-            DataStreamSource<TravelEvent> travelEventStream = env.addSource(travelEvent);
 
-           travelEventStream.print();
+            //旅行信息数据
+         //   FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
+            FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
+            DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
             //周期信号
-            FlinkKafkaConsumer<CycleSignalData> cycleSignalData =
-                    new FlinkKafkaConsumer<>("signal_cycle_data",
-                            new CycleSignalKafkaSchema(),
-                            kafkaProperties2);
+        //    FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
+            FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
             DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
 
-            cycleSignalDataStream.print();
-
 
             //2、使用connect，把2个流联合处理
-        ConnectedStreams<TravelEvent, CycleSignalData> connect = travelEventStream
+            ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
                     .connect(cycleSignalDataStream);
 
-        ConnectedStreams<TravelEvent, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
-                (KeySelector<TravelEvent, String>) travelInfo1 -> travelInfo1.getCrossId(),
+            ConnectedStreams<TravelInfo, CycleSignalData> travelInfoCycleSignalDataConnectedStreams = connect.keyBy(
+                    (KeySelector<TravelInfo, String>) travelInfo1 -> travelInfo1.getCrossId(),
                     (KeySelector<CycleSignalData, String>) cycleSignalData1 -> cycleSignalData1.getCrossCode()
             );
 
-        travelInfoCycleSignalDataConnectedStreams.process(new TravelEventAndCycleCoProcessFunction());
+            SingleOutputStreamOperator<List<TravelLineSinkInfo>> sinkDataStream = travelInfoCycleSignalDataConnectedStreams
+                    .process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
+
+            sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
 
-        env.execute();
+            env.execute("TravelSituationAnalysisJob");
 
         } catch (Exception e){
             e.printStackTrace();
         }
 
+
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/DirectionEvalSecondSink.java

-package com.zhht.irn.sink;
-
-import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
-import com.zhht.irn.utils.DateUtils;
-import com.zhht.irn.utils.DorisUtils;
-import com.zhht.irn.utils.DruidConnectPoolUtils;
-import com.zhht.irn.utils.MySQLUtils;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-
-import java.sql.Connection;
-import java.sql.Date;
-import java.sql.PreparedStatement;
-
-import static com.zhht.irn.consts.Constant.INSERT_BATCH_SIZE;
-
-/**
- * 按方向评价实时指标
- *
- * 结果表：app_direction_eval_second_metric
- * 计算结果指标包含 交通流量，排队长度，延误时间，停车次数
- */
-public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecondMetric> {
-    Connection connection;
-    PreparedStatement insertPstmt;
-
-    int counter = 0;
-
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        connection = DorisUtils.getConnection();
-        String insertSql="insert into app_direction_eval_second_metric" +
-                "(cross_id, lane_id, cycle_order, current_second, acc_arrived_flow, acc_queue_length, acc_stop_count," +
-                " stop_flag_count, statistic_time, cycle_start_time, cycle_end_time, record_date) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
-        insertPstmt = connection.prepareStatement(insertSql);
-    }
-    @Override
-    public void close() throws Exception {
-        super.close();
-        if (insertPstmt != null) insertPstmt.close();
-        if (connection != null) connection.close();
-    }
-
-    @Override
-    public void invoke(DirectionEvalSecondMetric metric, Context context) throws Exception {
-        insertPstmt.setString(1, metric.getCrossId());
-        insertPstmt.setString(2, metric.getLineId());
-        insertPstmt.setLong(3, metric.getCycleOrder());
-        insertPstmt.setLong(4, metric.getCurrentSecond());
-        insertPstmt.setInt(5, metric.getAccArrivedFlow());
-        insertPstmt.setInt(6, metric.getAccQueueLength());
-        insertPstmt.setInt(7, metric.getAccStopCount());
-        insertPstmt.setInt(8, metric.getStopFlagCount());
-        insertPstmt.setString(9, DateUtils.getTodayTime());
-        insertPstmt.setString(10, DateUtils.longToString(metric.getCycleStart(), DateUtils.timeFormat));
-        insertPstmt.setString(11, DateUtils.longToString(metric.getCycleEnd(), DateUtils.timeFormat));
-        insertPstmt.setDate(12, new Date(System.currentTimeMillis()));
-
-        if(counter < INSERT_BATCH_SIZE) {
-            counter ++;
-            insertPstmt.addBatch();
-        } else {
-            insertPstmt.executeBatch();
-            counter = 0;
-        }
-    }
-}

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java

@@ -52,6 +52,7 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
      */
     @Override
     public void invoke(List<LaneNorm> laneNorms, SinkFunction.Context context) throws Exception {
+
         if(laneNorms.size()==0){
             return;
         }

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelLaneInfoSink.java

+package com.zhht.irn.sink;
+
+import com.alibaba.druid.pool.DruidPooledConnection;
+import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
+import com.zhht.irn.utils.DorisUtils;
+import com.zhht.irn.utils.DruidConnectPoolUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.log4j.Logger;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * TravelLaneInfoSink
+ *
+ * @author Rui meng
+ * @description 行车情况信息sink
+ * @date 2022/12/15 9:39
+ */
+public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo>> {
+
+    private static Logger logger = Logger.getLogger(TravelLaneInfoSink.class);
+
+    Connection connection;
+
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        connection = DorisUtils.getConnection();
+        logger.info("建立数据库连接成功。。。。。。。");
+
+    }
+
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if(connection!=null){
+            connection.close();
+        }
+
+
+    }
+
+    @Override
+    public void invoke(List<TravelLineSinkInfo> value, Context context) throws Exception {
+        super.invoke(value, context);
+
+        //重新获取连接，避免超时的问题
+        connection = DorisUtils.getConnection();
+        String sql = "INSERT INTO app_cross_line_travel_car_info" +
+                "(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
+                " flow_direction, pass_numbers," +
+                " last_car_inCross_time, average_pass_time, average_pass_speed, average_control_delay, " +
+                "average_stop_delay, average_stop_times, average_car_head_time_gap" +
+                ")" +
+                "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?,?,?,?) ";
+        PreparedStatement preparedStatement = connection.prepareStatement(sql);
+
+        //TODO 这里的时间使用数据携带的时间，使用最后进入路口车辆的时间，这样在补数据时，可以补充到对应的分区下
+        Date date = new Date();
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
+        String statistic_time = sdf.format(date);
+
+
+        for (TravelLineSinkInfo t :value) {
+            // 使用事件数据的时间划分数据分区，这样可以进行补数操作
+            String record_date = sdf2.format(t.getLast_car_inCross_time());
+
+            preparedStatement.setString(1, record_date);
+            preparedStatement.setString(2, t.getCross_id());
+            preparedStatement.setString(3, statistic_time);
+            preparedStatement.setString(4, sdf.format(t.getCycle_begin_time()));
+            preparedStatement.setString(5, sdf.format(t.getCycle_end_time()));
+            preparedStatement.setString(6, t.getLane_id());
+            preparedStatement.setString(7, t.getCycle_id().toString());
+            preparedStatement.setString(8, t.getDirection());
+            preparedStatement.setString(9, t.getFlow_direction());
+            preparedStatement.setInt(10, t.getPass_numbers());
+            preparedStatement.setLong(11, t.getLast_car_inCross_time());
+            preparedStatement.setDouble(12, t.getAverage_pass_time());
+            preparedStatement.setDouble(13, t.getAverage_pass_speed());
+            preparedStatement.setDouble(14, t.getAverage_control_delay());
+            preparedStatement.setDouble(15, t.getAverage_stop_delay());
+            preparedStatement.setDouble(16, t.getAverage_stop_times());
+            preparedStatement.setDouble(17, t.getAverage_car_head_time_gap());
+            preparedStatement.addBatch();
+        }
+        int[] ints = preparedStatement.executeBatch();
+
+        logger.info("写入数据成功。。。。。"+ints.length+"...");
+        preparedStatement.close();
+
+        connection.close();
+
+
+    }
+}

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

@@ -1038,6 +1038,15 @@ public class DateUtils {
         } catch(Exception e) { e.printStackTrace();}
         return longToString(time, dateFormat);
     }
+
+    public static String longToStandardFormatString(String timestamp) {
+        return longToString(timestamp, "yyyy-MM-dd HH:mm:ss");
+    }
+
+    public static String longToStandardFormatString(long timestamp) {
+        return longToString(timestamp, "yyyy-MM-dd HH:mm:ss");
+    }
+
     /**
      * 将字符串日期转换为日期格式
      * 自定義格式

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

 package com.zhht.irn.utils;
 
+import org.apache.doris.flink.cfg.DorisExecutionOptions;
+import org.apache.doris.flink.cfg.DorisOptions;
+import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.sink.DorisSink;
+import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
 import org.apache.flink.api.java.utils.ParameterTool;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -44,4 +50,34 @@ public class DorisUtils {
             }
         }
     }
+
+    public static DorisSink.Builder<String> getBuilder(String filePath) throws IOException {
+        DorisSink.Builder<String> builder = DorisSink.builder();
+        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
+
+        InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream(filePath);
+        ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
+
+        // 根据路径的配置文件获取MySQL链接参数
+        String username = tool.getRequired("username");
+        String password = tool.getRequired("password");
+        String host = tool.getRequired("host");
+        String port = tool.getRequired("port");
+        String database = tool.get("database", "default");
+        String table = tool.getRequired("table");
+        String label = tool.getRequired("label");
+        dorisBuilder.setFenodes(host + ":" + port)
+                .setTableIdentifier(database + "." + table)
+                .setUsername(username)
+                .setPassword(password);
+
+        DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
+        executionBuilder.setLabelPrefix(label); // streamload label prefix
+
+        builder.setDorisReadOptions(DorisReadOptions.builder().build())
+                .setDorisExecutionOptions(executionBuilder.build())
+                .setSerializer(new SimpleStringSerializer()) //serialize according to string
+                .setDorisOptions(dorisBuilder.build());
+        return builder;
+    }
 }

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

@@ -31,8 +31,8 @@ public class DruidConnectPoolUtils {
         dataSource.setUrl(dbHosts);
         dataSource.setUsername(username);
         dataSource.setPassword(password);
-        dataSource.setInitialSize(3);  //初始连接数，默认0
-        dataSource.setMaxActive(10);  //最大连接数，默认8
+        dataSource.setInitialSize(5);  //初始连接数，默认0
+        dataSource.setMaxActive(30);  //最大连接数，默认8
         dataSource.setMinIdle(5);  //最小闲置数
         dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
         dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
@@ -47,8 +47,8 @@ public class DruidConnectPoolUtils {
         dataSource.setUrl(dbHosts);
         dataSource.setUsername(username);
         dataSource.setPassword(password);
-        dataSource.setInitialSize(3);  //初始连接数，默认0
-        dataSource.setMaxActive(10);  //最大连接数，默认8
+        dataSource.setInitialSize(5);  //初始连接数，默认0
+        dataSource.setMaxActive(30);  //最大连接数，默认8
         dataSource.setMinIdle(5);  //最小闲置数
         dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
         dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除

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

@@ -152,10 +152,45 @@ public class FlinkUtils {
         env.enableCheckpointing(checkpointInterval);
         env.setStateBackend(new FsStateBackend(checkpointPath));
         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
-        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
+        /**
+         * 暂时不需要失败自动重试
+         */
+        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
+
+        // 基于Kafka 原始JSON格式的字符串，获取消息流
+        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
+        DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
+        return rawStringStream;
+    }
+
+    public static DataStream<String> createKafkaStreamHuangzhun(String[] args, StreamExecutionEnvironment environment, String topics, String groupId) {
+        ParameterTool kafkaTool = ParameterTool.fromArgs(args);
+        String servers = kafkaTool.get("bootstrap.servers", defaultBootstrapServers);
+        String topics2 = kafkaTool.get("kafka.input.topics", topics);
+        String groupId2 = kafkaTool.get("group.id", groupId);
+        String autoOffsetReset = kafkaTool.get("auto.offset.reset", defaultAutoOffsetReset);
+        String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
+        long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
+        String enableAutoCommit = kafkaTool.get("enable.auto.commit", "false");
+
+        return buildKafkaDataStreamHuangzhun(environment, servers, topics2, groupId2, autoOffsetReset, checkpointPath, checkpointInterval, enableAutoCommit);
+    }
 
+    private static DataStream<String> buildKafkaDataStreamHuangzhun(StreamExecutionEnvironment env, String servers, String topics, String groupId,
+                                                           String autoOffsetReset, String checkpointPath, long checkpointInterval, String enableAutoCommit) {
+        List<String> topicList = Arrays.asList(topics.split(","));
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", servers);
+        properties.setProperty("group.id", groupId);
+        properties.setProperty("enable.auto.commit", enableAutoCommit);
+        properties.setProperty("auto.offset.reset", autoOffsetReset);
+        env.enableCheckpointing(checkpointInterval);
+        env.setStateBackend(new FsStateBackend(checkpointPath));
+        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
         // 基于Kafka 原始JSON格式的字符串，获取消息流
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
+        kafkaConsumer.setStartFromLatest();
         DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
         return rawStringStream;
     }

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

+package com.zhht.irn.utils;
+
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.DriverManager;
+
+/**
+ * @Description:
+ * @Author ：Marinh
+ * @Param:
+ * @retrun:
+ * @Creat :2023-01-06-12:32
+ **/
+public class applicationUtils {
+    public static String getLoggerLevel() throws Exception {
+        InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("application.properties");
+        ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
+        // 根据路径的配置文件获取MySQL链接参数
+        String driver = tool.getRequired("log.info.level");
+        return driver;
+    }
+
+}

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

+log.info.level=info

realtime/hologram-streaming/src/main/resources/doris-connector-travel-info.properties

+host=10.243.0.22
+port=8030
+username=root
+password=mima
+database=qxlk
+table=app_travel_metric
+label=app_travel_metric-import
\ No newline at end of file

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

+host=10.243.0.22
+port=8030
+username=root
+password=mima
+database=qxlk
+table=app_direction_eval_second_metric
+label=app_direction_eval_second_metric-import
\ No newline at end of file

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

 driver=com.mysql.jdbc.Driver
-host=10.243.0.22
+host=localhost
 port=9030
 username=root
 password=mima

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

@@ -4,4 +4,4 @@ auto.offset.reset=latest
 group.id=TravelInfoJob
 checkpoint.path=file:///data/flink/state
 checkpoint.interval=60000
-enable.auto.commit=false
\ No newline at end of file
+enable.auto.commit=true
\ No newline at end of file