调整逻辑，指标结果需要存储对应的周期开始和结束时间

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

@@ -15,4 +15,6 @@ public class Constant {
 
     // 周期计算的延迟时间 15秒
     public static final int CYCLE_CALCULATE_LATENESS = 15 * 1000;
+
+    public static final int INSERT_BATCH_SIZE = 1000;
 }

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

@@ -10,6 +10,10 @@ public class DirectionEvalSecondMetric {
     private String crossId;
     private String lineId;
     private long cycleOrder;
+
+    private long cycleStart;
+
+    private long cycleEnd;
     private long currentSecond;
 
     private int accArrivedFlow;
@@ -41,6 +45,22 @@ public class DirectionEvalSecondMetric {
         this.cycleOrder = cycleOrder;
     }
 
+    public long getCycleStart() {
+        return cycleStart;
+    }
+
+    public void setCycleStart(long cycleStart) {
+        this.cycleStart = cycleStart;
+    }
+
+    public long getCycleEnd() {
+        return cycleEnd;
+    }
+
+    public void setCycleEnd(long cycleEnd) {
+        this.cycleEnd = cycleEnd;
+    }
+
     public long getCurrentSecond() {
         return currentSecond;
     }
@@ -84,9 +104,11 @@ public class DirectionEvalSecondMetric {
     @Override
     public String toString() {
         return "DirectionEvalSecondMetric{" +
-                "crossId='" + crossId + '\'' +
-                ", lineId='" + lineId + '\'' +
+                "crossId=" + crossId +
+                ", lineId=" + lineId +
                 ", cycleOrder=" + cycleOrder +
+                ", cycleStart=" + cycleStart +
+                ", cycleEnd=" + cycleEnd +
                 ", currentSecond=" + currentSecond +
                 ", accArrivedFlow=" + accArrivedFlow +
                 ", accQueueLength=" + accQueueLength +

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

@@ -140,6 +140,8 @@ public class CalDirectionSecondMetricFunction extends ProcessJoinFunction<Cycle,
 
 
                         DirectionEvalSecondMetric metric = new DirectionEvalSecondMetric();
+                        metric.setCycleStart(cycleStart);
+                        metric.setCycleEnd(cycleEnd);
                         metric.setCrossId(cycle.getCrossCode());
                         metric.setLineId(entry.getKey());
                         metric.setCurrentSecond(start);

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

@@ -6,7 +6,6 @@ import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.TravelEvent;
 import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
 import com.zhht.irn.sink.DirectionEvalSecondSink;
-import com.zhht.irn.utils.FileUtils;
 import com.zhht.irn.utils.FlinkUtils;
 import com.zhht.irn.utils.StringUtils;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
@@ -36,10 +35,8 @@ public class DirectionEvalJob {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
         // 获取信控数据-周期结束实时流 Kafka
-
         DataStream<Cycle> cycleStream =
-//                env.addSource(new CycleMockSource())
-                FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "directionEvalJob")
+                FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
                     System.out.println("信控周期->" + record);
                     return JSON.parseObject(record, Cycle.class);
@@ -47,8 +44,7 @@ public class DirectionEvalJob {
 
         // 获取旅行事件数据实时流 kafka
         DataStream<TravelEvent> travelEventStream =
-//                env.addSource(new TravelEventMockSource())
-                FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME)
+                FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
                 .map(record -> {
                     System.out.println("旅行事件->" + record);
                     return JSON.parseObject(record, TravelEvent.class);
@@ -66,7 +62,6 @@ public class DirectionEvalJob {
                 .withTimestampAssigner((xx, timestamp) -> Long.parseLong(xx.getEventTime()));
 
         // 周期数据 interval join 事件流数据，计算 交通流量和排队长度两个指标
-
         // 旅行事件数据流
         KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream.assignTimestampsAndWatermarks(travelEventWatermarkStrategy)
                 .keyBy(record -> record.getCrossId());

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

 package com.zhht.irn.job;
 
 import com.alibaba.fastjson.JSON;
+import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.sink.TravelMetricSink;
@@ -30,10 +31,10 @@ public class TravelInfoJob {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
         // 通过kafka配置文件，获取String类型的实时数据流
-        DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env);
-//        DataStream<String> rawStringStream = env.readTextFile("./data/2022-11-23--11/.part-820a21c1-f4fa-485e-a3ed-b65e15fefdfa-0.inprogress.59c1cbee-8597-4dd7-a2fc-12c8df8db83d");
-        // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查
-        rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
+        DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_TOPIC_NAME, "TravelInfoJob");
+
+        // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查，需要时打开
+        // rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
 
         // 开始计算旅行信息相关指标
         rawStringStream.map((MapFunction<String, Travel>) s -> {

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.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;
+
 /**
  * 按方向评价实时指标
  *
@@ -18,15 +23,16 @@ public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecon
     Connection connection;
     PreparedStatement insertPstmt;
 
+    int counter = 0;
+
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = MySQLUtils.getConnection();
+        connection = DorisUtils.getConnection();
         String insertSql="insert into app_direction_eval_second_metric" +
-                "(cross_id, line_id, cycle_order, current_second, acc_arrived_flow, acc_queue_length, acc_stop_count, stop_flag_count) values (?, ?, ?, ?, ?, ?, ?, ?)";
-
+                "(cross_id, lane_id, cycle_order, current_second, acc_arrived_flow, acc_queue_length, acc_stop_count," +
+                " stop_flag_count, statistic_time, cycle_start, cycle_end) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         insertPstmt = connection.prepareStatement(insertSql);
-
     }
     @Override
     public void close() throws Exception {
@@ -46,6 +52,16 @@ public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecon
         insertPstmt.setInt(6, metric.getAccQueueLength());
         insertPstmt.setInt(7, metric.getAccStopCount());
         insertPstmt.setInt(8, metric.getStopFlagCount());
-        insertPstmt.execute();
+        insertPstmt.setDate(9, new Date(System.currentTimeMillis()));
+        insertPstmt.setString(10, DateUtils.longToString(metric.getCycleStart(), DateUtils.timeFormat));
+        insertPstmt.setString(11, DateUtils.longToString(metric.getCycleEnd(), DateUtils.timeFormat));
+
+        if(counter < INSERT_BATCH_SIZE) {
+            counter ++;
+            insertPstmt.addBatch();
+        } else {
+            insertPstmt.executeBatch();
+            counter = 0;
+        }
     }
 }

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

@@ -3,4 +3,4 @@ host=127.0.0.1
 port=9030
 username=root
 password=mima
-database=demo
+database=qxlk