Merge remote-tracking branch 'origin/master'

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

+package com.zhht.irn.functions;
+
+import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.util.Collector;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * TravelLineSinkInfoFlatMapFunction
+ *
+ * @author Rui meng
+ * @description 行车情况切分为字符串
+ * @date 2023/1/9 16:42
+ */
+public class TravelLineSinkInfoFlatMapFunction implements FlatMapFunction<List<TravelLineSinkInfo>, String> {
+
+    @Override
+    public void flatMap(List<TravelLineSinkInfo> travelLineSinkInfos, Collector<String> collector) throws Exception {
+
+        //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:travelLineSinkInfos) {
+            // 使用事件数据的时间划分数据分区，这样可以进行补数操作
+           String record_date = sdf2.format(t.getLast_car_inCross_time());
+          //  String record_date = "2023-01-09";
+
+           collector.collect(record_date+"\t"+t.getCross_id()+"\t"+statistic_time+"\t"+t.getCycle_begin_time()+"\t"+t.getCycle_end_time()
+           +"\t"+t.getLane_id()+"\t"+t.getCycle_id()+"\t"+t.getDirection()+"\t"+t.getFlow_direction()+"\t"+t.getPass_numbers()
+           +"\t"+t.getLast_car_inCross_time()+"\t"+t.getAverage_pass_time()+"\t"+t.getAverage_pass_speed()+"\t"+t.getAverage_control_delay()
+           +"\t"+t.getAverage_stop_delay()+"\t"+t.getAverage_stop_times()+"\t"+ t.getAverage_car_head_time_gap());
+
+        }
+    }
+}

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

@@ -2,17 +2,22 @@ package com.zhht.irn.job;
 
 import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.*;
+import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
 import com.zhht.irn.enums.EventType;
 import com.zhht.irn.functions.LaneNormProcessFunction;
 import com.zhht.irn.sink.LaneNormSink;
+import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.DruidConnectPoolUtils;
 import com.zhht.irn.utils.FlinkUtils;
+import org.apache.doris.flink.sink.DorisSink;
 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.api.java.utils.ParameterTool;
 import org.apache.flink.streaming.api.datastream.*;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
@@ -20,6 +25,8 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import java.time.Duration;
 import java.util.*;
 
+import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
+
 /**
  * 两阶段提交参考文档：
  * https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
@@ -30,11 +37,16 @@ public class LaneNormJob {
     public static void main(String[] args) throws Exception {
         try {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+            ParameterTool kafkaTool = ParameterTool.fromArgs(args);
+            String label = kafkaTool.get("label");
             //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");
+            //测试用topic
+            //DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"s_data2","laneNormJob1");
+            //DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"t_info2","laneNormJob1");
             KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
                 @Override
                 public String getKey(Cycle cycle) throws Exception {
@@ -79,12 +91,23 @@ public class LaneNormJob {
             //根据corssid 分组 保证同一路口的数据进行一个进程
             DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy).keyBy(cycleKeySelect);
             DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).keyBy(carKeySelector).filter(filterFunction);
-
             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(" ");
+            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-lane-norm.properties",label);
+            cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4)
+                    .map((MapFunction<LaneNorm, String>) desm -> {
+                // 将结果实体类，拼接成制表符分割的字符串，进行sink，要保持字段顺序跟表的字段顺序相一致
+                return desm.getCorssId() + "\t" + desm.getLaneId() + "\t" + desm.getDirection() + "\t" + desm.getCycleOrder() + "\t"
+                        + desm.getLaneName() + "\t" + desm.getOrder() + "\t" + desm.getCycleStartTime() + "\t"
+                        + desm.getCycleEndTime()+ "\t" + desm.getMaxQueueLength() + "\t" + desm.getEffectGreenTime() + "\t"
+                        + desm.getPassCapacity()+ "\t" + desm.getTrafficCapacity() + "\t" + desm.getSaturation() + "\t"
+                        + desm.getResidualCapacity()+ "\t" + desm.getSpotSpaceShare() + "\t"
+                        + DateUtils.getTodayTime() + "\t" + longToStandardFormatString(System.currentTimeMillis())+ "\t"
+                        + desm.getType()+ "\t" + desm.getSpotSpaceLength();
+            }).sinkTo(builder.build());
+            //process.addSink(new LaneNormSink()).name(" ");
             env.execute("laneNormJob");
         }catch (Exception e){
+            e.printStackTrace();
             System.out.println("程序异常关闭->"+e);
         }
 

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

@@ -6,11 +6,14 @@ 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.functions.TravelLineSinkInfoFlatMapFunction;
 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.doris.flink.sink.DorisSink;
+import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.common.time.Time;
@@ -24,6 +27,7 @@ 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.util.Collector;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
 
@@ -65,7 +69,7 @@ public class TravelSituationAnalysisJob {
             env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
             //设置重启策略 这里不设置重启策略，由平台统一拉起任务
-            // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
+            //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
             env.setRestartStrategy(RestartStrategies.noRestart());
 
 
@@ -90,16 +94,16 @@ public class TravelSituationAnalysisJob {
 
 
             //旅行信息数据
-         //   FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
-            FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
+          // 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<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
-            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);
 
-
             //2、使用connect，把2个流联合处理
             ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
                     .connect(cycleSignalDataStream);
@@ -112,7 +116,13 @@ public class TravelSituationAnalysisJob {
             SingleOutputStreamOperator<List<TravelLineSinkInfo>> sinkDataStream = travelInfoCycleSignalDataConnectedStreams
                     .process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
 
-            sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
+           // sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
+
+
+            // 获取builder
+            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-travel-situation.properties");
+
+            sinkDataStream.flatMap(new TravelLineSinkInfoFlatMapFunction()).sinkTo(builder.build());
 
             env.execute("TravelSituationAnalysisJob");
 

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

@@ -52,12 +52,6 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
         if(laneNorms.size()==0){
             return;
         }
-        /**
-         * 添加连接判断 防止jdbc 失效
-         */
-        if(!DataSourceUtils.isDbConnected(connection)){
-            connection=DorisUtils.getConnection();
-        }
         for (LaneNorm value: laneNorms){
             insertPstmt.setString(1, value.getCorssId());
             insertPstmt.setString(2, value.getLaneId());

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

@@ -12,10 +12,10 @@ import java.sql.SQLException;
  **/
 public class DataSourceUtils {
 
-    public static boolean isDbConnected(Connection con) {
+    public static boolean  isDbConnected(Connection con) {
         //final String CHECK_SQL_QUERY = "SELECT 1";
         try {
-            if(con != null && !con.isValid(1000)){
+            if(con != null && con.isValid(1000)){
                 return true;
             }
         } catch (SQLException e) {

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

@@ -83,4 +83,33 @@ public class DorisUtils {
                 .setDorisOptions(dorisBuilder.build());
         return builder;
     }
+
+    public static DorisSink.Builder<String> getBuilder(String filePath,String label1) 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(StringUtils.isNotEmpty(label1)?label1: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/FlinkUtils.java

@@ -156,7 +156,7 @@ public class FlinkUtils {
          * 暂时不需要失败自动重试
          */
         //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
-
+        env.setRestartStrategy(RestartStrategies.noRestart());
         // 基于Kafka 原始JSON格式的字符串，获取消息流
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
         DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
@@ -187,7 +187,7 @@ 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.noRestart());
         // 基于Kafka 原始JSON格式的字符串，获取消息流
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
         kafkaConsumer.setStartFromLatest();

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

+host=10.243.0.22
+port=8030
+username=root
+password=mima
+database=qxlk
+table=app_cross_dimension_norm
+label=lanenorm-import

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

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