Merge remote-tracking branch 'origin/master'

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

@@ -30,7 +30,7 @@ import java.util.stream.Stream;
  * @retrun:
  * @Creat :2022-11-04-9:53
  **/
-public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>> {
+public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm> {
     // state 管理当前周期中存储的旅行事件
     private MapState<String, List<TravelEvent>> cycleTravel;
     // 临时
@@ -93,7 +93,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     }
 
     @Override
-    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
+    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
         //状态位修改为一个路口一个状态位
         if(!processCycle.contains(cycle.getCrossCode())){
             processCycle.put(cycle.getCrossCode(),false);
@@ -185,10 +185,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         getPositionNorm(list, positionList, f1, laneNorms);
         //获取所有流向的指标
         getPositionNameNorm(list, positionNameList, f1, laneNorms,loggerLevel);
-
-        String afterTimeSecond = DateUtils.getAfterTimeSecond(20);
         if(laneNorms.size()>0){
-            out.collect(laneNorms);
+            for(LaneNorm ln:laneNorms){
+                out.collect(ln);
+            }
         }
         //本次计算中未发生进入事件的数据
         List<TravelEvent> collect = car.stream().filter(recode -> StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList());
@@ -205,7 +205,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     }
 
     @Override
-    public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
+    public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
         if(!processCycle.contains(event.getCrossId())){
             processCycle.put(event.getCrossId(),false);
         }
@@ -232,7 +232,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     }
 
     @Override
-    public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<LaneNorm>> out) throws Exception {
+    public void onTimer(long timestamp, OnTimerContext ctx, Collector<LaneNorm> out) throws Exception {
         super.onTimer(timestamp, ctx, out);
         //System.out.println("触发定时器了。。。。");
         //通过定时器清空字典表 防止字典表数据有更新
@@ -316,6 +316,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                 String position = String.valueOf(map.get("position"));
                 String name = String.valueOf(map.get("name"));
                 LaneNorm laneNorm = new LaneNorm();
+                laneNorm.setLaneId("0");
                 laneNorm.setLaneName(position + name);
                 laneNorm.setOrder(0);
                 laneNorm.setCycleOrder(cycle.getCycleOrder());
@@ -369,6 +370,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
             for (Map map : positionList) {
                 String position = String.valueOf(map.get("position"));
                 LaneNorm laneNorm = new LaneNorm();
+                laneNorm.setLaneId("0");
                 laneNorm.setLaneName("整体");
                 laneNorm.setOrder(1);
                 laneNorm.setCycleOrder(cycle.getCycleOrder());

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,6 +37,8 @@ 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)));
@@ -83,8 +92,19 @@ public class LaneNormJob {
             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();

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

@@ -80,4 +80,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-info.properties

@@ -4,4 +4,5 @@ username=root
 password=mima
 database=qxlk
 table=app_travel_metric
-label=app_travel_metric-import
+label=app_travel_metric-import
\ No newline at end of file
+label-lanenorm=lanenorm