车道任务和行车情况任务调整下flink-connector的写入写法，同时删除不必要的配置文件

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

@@ -2,20 +2,15 @@ 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.*;
@@ -37,8 +32,6 @@ 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)));
@@ -92,7 +85,7 @@ 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());
-            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-lane-norm.properties",label);
+            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties",args, "app_cross_dimension_norm");
             cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4)
                     .map((MapFunction<LaneNorm, String>) desm -> {
                 // 将结果实体类，拼接成制表符分割的字符串，进行sink，要保持字段顺序跟表的字段顺序相一致

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

 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.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;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
@@ -27,7 +20,6 @@ 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;
 
 
@@ -45,12 +37,11 @@ public class TravelSituationAnalysisJob {
     public static void main(String[] args)  {
 
         try {
-
             InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("kafka.properties");
             ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
 
 
-//1、创建环境
+            //1、创建环境
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             // checkpoint配置
             CheckpointConfig checkpointConfig = env.getCheckpointConfig();
@@ -63,7 +54,7 @@ public class TravelSituationAnalysisJob {
             checkpointConfig.setMaxConcurrentCheckpoints(1);
 
             //这如果不指定使用FsStateBackend 则默认使用flink集群中的配置，rocksDB
-         //   env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
+            //   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);
@@ -120,7 +111,7 @@ public class TravelSituationAnalysisJob {
 
 
             // 获取builder
-            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-travel-situation.properties");
+            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_cross_line_travel_car_info");
 
             sinkDataStream.flatMap(new TravelLineSinkInfoFlatMapFunction()).sinkTo(builder.build());
 

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

-package com.zhht.irn.sink;
-
-import com.zhht.irn.entity.LaneNorm;
-import com.zhht.irn.utils.*;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.util.List;
-
-/**
- * @Description:
- * @Author ：Marinh
- * @Param:
- * @retrun:
- * @Creat :2022-10-31-14:11
- **/
-public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
-
-    Connection connection;
-    PreparedStatement insertPstmt;
-    PreparedStatement updatePstmt;
-
-
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        // path暂定为固定值
-        connection = DorisUtils.getConnection();
-        insertPstmt = connection.prepareStatement("insert into " +
-                "app_cross_dimension_norm(cross_id,lane_id,direction,cycle_order,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," +
-                "spot_space_share,dimension_name,`order`,statistic_time,`type`,spot_space_length,cycle_start_time,cycle_end_time,record_date) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
-    }
-
-    @Override
-    public void close() throws Exception {
-        super.close();
-        if(insertPstmt != null) insertPstmt.close();
-        if(connection != null) connection.close();
-    }
-
-    /**
-     * 来一条数据就执行一次
-     *
-     * 1000w的数据  1000w次
-     */
-    @Override
-    public void invoke(List<LaneNorm> laneNorms, SinkFunction.Context context) throws Exception {
-        if(laneNorms.size()==0){
-            return;
-        }
-        for (LaneNorm value: laneNorms){
-            insertPstmt.setString(1, value.getCorssId());
-            insertPstmt.setString(2, value.getLaneId());
-            insertPstmt.setString(3, value.getDirection());
-            insertPstmt.setLong(4, value.getCycleOrder());
-            insertPstmt.setDouble(5, value.getMaxQueueLength());
-            insertPstmt.setString(6,value.getEffectGreenTime());
-            insertPstmt.setDouble(7,value.getPassCapacity());
-            insertPstmt.setDouble(8,value.getTrafficCapacity());
-            insertPstmt.setDouble(9,value.getSaturation());
-            insertPstmt.setDouble(10,value.getResidualCapacity());
-            insertPstmt.setDouble(11,value.getSpotSpaceShare());
-            insertPstmt.setString(12,value.getLaneName());
-            insertPstmt.setInt(13,value.getOrder());
-            insertPstmt.setString(14, DateUtils.getTodayTime());
-            insertPstmt.setString(15,value.getType());
-            insertPstmt.setInt(16,value.getSpotSpaceLength());
-            insertPstmt.setString(17,value.getCycleStartTime());
-            insertPstmt.setString(18,value.getCycleEndTime());
-            insertPstmt.setString(19,DateUtils.getTodayTime());
-            insertPstmt.addBatch();
-        }
-        insertPstmt.executeBatch();
-    }
-}

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/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