将按方向评价实时任务改造为flink-doris-connector的方式进行插入。

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 = 1000;
 }

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/job/DirectionEvalJob.java

@@ -4,17 +4,27 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.consts.Constant;
 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.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.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.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.utils.ParameterTool;
 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.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.InputStream;
 import java.time.Duration;
 
 /**
@@ -63,12 +73,45 @@ public class DirectionEvalJob {
         KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream.assignTimestampsAndWatermarks(travelEventWatermarkStrategy)
                 .keyBy(record -> record.getCrossId());
 
+        // 定义好doris sink
+        DorisSink.Builder<String> builder = DorisSink.builder();
+        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
+
+        InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("doris-connector.properties");
+        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.get("table", "app_direction_eval_second_metric");
+        String label = tool.get("label", "app_direction_eval_second_metric-import");
+        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());
+
         // 周期数据join 旅行事件
         cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
                 .keyBy(record -> record.getCrossCode())
                 .connect(travelEventKeyedStream)
                 .process(new CalDirectionSecondMetricFunction())
-                .addSink(new DirectionEvalSecondSink()).setParallelism(4);
+                .map((MapFunction<DirectionEvalSecondMetric, String>) desm -> {
+                    return desm.getCrossId() + "\t" + desm.getLineId() + "\t" + desm.getCycleOrder() + "\t" + DateUtils.longToString(desm.getCycleStart(), "yyyy-MM-dd HH:mm:ss") + "\t" + DateUtils.longToString(desm.getCycleEnd(), "yyyy-MM-dd HH:mm:ss") + "\t"
+                            + desm.getCurrentSecond() + "\t" + desm.getAccArrivedFlow() + "\t" + desm.getAccQueueLength() + "\t" + desm.getStopFlagCount() + "\t"
+                            + desm.getAccStopCount() + "\t" + DateUtils.getCurrentDateTime("yyyy-MM-dd HH:mm:ss") + "\t" + DateUtils.getCurrentDateTime("yyyy-MM-dd");
+                }).sinkTo(builder.build());
 
         // 旅行数据流
         env.execute("DirectionEvalJob");

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