提取builder公共代码，封装到DorisUtils中

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

@@ -10,23 +10,19 @@ 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;
 
+import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
+
 /**
  * 按方向评价实时任务，按秒计算
  *
@@ -73,34 +69,8 @@ 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());
+        // 定义好doris sink builder
+        DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector.properties");
 
         // 周期数据join 旅行事件
         cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
@@ -108,9 +78,11 @@ public class DirectionEvalJob {
                 .connect(travelEventKeyedStream)
                 .process(new CalDirectionSecondMetricFunction())
                 .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");
+                    // 将结果实体类，拼接成制表符分割的字符串，进行sink，要保持字段顺序跟表的字段顺序相一致
+                    return desm.getCrossId() + "\t" + desm.getLineId() + "\t" + desm.getCycleOrder() + "\t" + longToStandardFormatString(desm.getCycleStart()) + "\t"
+                            + longToStandardFormatString(desm.getCycleEnd()) + "\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());
 
         // 旅行数据流

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

@@ -5,7 +5,10 @@ import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.sink.TravelMetricSink;
+import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.FlinkUtils;
+import org.apache.doris.flink.sink.DorisSink;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -15,6 +18,7 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 
 import static com.zhht.irn.utils.StringUtils.isNotEmpty;
+import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
 
 /**
  * 旅行信息实时任务
@@ -33,7 +37,10 @@ public class TravelInfoJob {
         DataStream<String> rawStringStream = FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_TOPIC_NAME, "TravelInfoJob");
 
         // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查，需要时打开
-        // rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
+        // rawStringStream.sinkTo(FileUtils.getFileSink("./data/
+
+        // 获取builder
+        DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-travel-info.properties");
 
         DataStream<Travel> travelStream = rawStringStream
                 // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
@@ -124,7 +131,19 @@ public class TravelInfoJob {
                     "停车总时长：{}", travel.getRecordId(), maxSpeed, minSpeed, travel.getAvgSpeed(), stopCount, stopTime);
 
             return travel;
-        }).addSink(new TravelMetricSink());
+        }).map((MapFunction<Travel, String>) travel -> {
+            return travel.getRecordId() + "\t" + travel.getCrossId() + "\t" + longToStandardFormatString(travel.getRecordTime()) + "\t"
+                    + longToStandardFormatString(travel.getTravelBeginTime()) + "\t" + longToStandardFormatString(travel.getTravelEndTime()) + "\t"
+                    + travel.getFlowDirection() + "\t" + travel.getCarId() + "\t" + travel.getCarColor() + "\t" + travel.getPlate() + "\t"
+                    + travel.getPlateColor() + "\t" + travel.getCarType() + "\t" + travel.getFirstLineId() + "\t" + longToStandardFormatString(travel.getFirstTime()) + "\t"
+                    + travel.getFirstSpeed() + "\t" + travel.getArrivedLineId() + "\t" + longToStandardFormatString(travel.getArrivedTime()) + "\t"
+                    + travel.getArrivedSpeed() + "\t" + travel.getInCrossLineId() + "\t" + longToStandardFormatString(travel.getInCrossTime()) + "\t"
+                    + travel.getInSpeed() + "\t" + travel.getOutCrossLineId() + "\t" + longToStandardFormatString(travel.getOutCrossTime()) + "\t"
+                    + travel.getOutSpeed() + "\t" + travel.getAwayLineId() + "\t" + longToStandardFormatString(travel.getAwayTime()) + "\t"
+                    + travel.getAwaySpeed() + "\t" + travel.getLastLineId() + "\t" + longToStandardFormatString(travel.getLastTime()) + "\t"
+                    + travel.getLastSpeed() + "\t" + travel.getCrossingTime() + "\t" + travel.getMinSpeed() + "\t" + travel.getMaxSpeed() + "\t"
+                    + travel.getAvgSpeed() + "\t" + travel.getStopCount() + "\t" + travel.getStopTime() + "\t" + longToStandardFormatString(System.currentTimeMillis());
+        }).sinkTo(builder.build());
 
         env.execute("TravelInfoJob");
     }

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

@@ -1038,6 +1038,15 @@ public class DateUtils {
         } catch(Exception e) { e.printStackTrace();}
         return longToString(time, dateFormat);
     }
+
+    public static String longToStandardFormatString(String timestamp) {
+        return longToString(timestamp, "yyyy-MM-dd HH:mm:ss");
+    }
+
+    public static String longToStandardFormatString(long timestamp) {
+        return longToString(timestamp, "yyyy-MM-dd HH:mm:ss");
+    }
+
     /**
      * 将字符串日期转换为日期格式
      * 自定義格式

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

 package com.zhht.irn.utils;
 
+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.java.utils.ParameterTool;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -44,4 +50,34 @@ public class DorisUtils {
             }
         }
     }
+
+    public static DorisSink.Builder<String> getBuilder(String filePath) 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(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/resources/doris-connector-travel-info.properties

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