将flink-doris-connector改成按字段写入

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

+package com.zhht.irn.functions;
+
+import com.zhht.irn.entity.Travel;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import java.time.LocalDate;
+import java.util.Calendar;
+import java.util.Date;
+
+import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
+
+/**
+ * Travel2RowDataMapFunction
+ *
+ * @author 吴延飞
+ * @description 将旅行信息实体类转化为
+ * @date 2023/1/9 16:42
+ */
+public class Travel2RowDataMapFunction extends RichMapFunction<Travel, RowData> {
+    private static Calendar calendar;
+    @Override
+    public void open(Configuration parameters) {
+        calendar = Calendar.getInstance();
+    }
+
+
+    @Override
+    public RowData map(Travel travel) throws Exception {
+        long recordTime = Long.valueOf(travel.getRecordTime());
+        Date recordDate = new Date(recordTime);
+        calendar.setTime(recordDate);
+        LocalDate localDate = LocalDate.ofYearDay(calendar.get(Calendar.YEAR), calendar.get(Calendar.DAY_OF_YEAR));
+        GenericRowData genericRowData = new GenericRowData(36);
+        genericRowData.setField(0, StringData.fromString(travel.getRecordId()));
+        genericRowData.setField(1, StringData.fromString(travel.getCrossId()));
+        genericRowData.setField(2, (int)localDate.toEpochDay());
+
+        genericRowData.setField(3, StringData.fromString(longToStandardFormatString(travel.getTravelBeginTime())));
+        genericRowData.setField(4, StringData.fromString(longToStandardFormatString(travel.getTravelEndTime())));
+        genericRowData.setField(5, StringData.fromString(travel.getFlowDirection()));
+        genericRowData.setField(6, StringData.fromString(travel.getCarId() + ""));
+        genericRowData.setField(7, StringData.fromString(travel.getCarColor()));
+        genericRowData.setField(8, StringData.fromString(travel.getPlate()));
+        genericRowData.setField(9, StringData.fromString(travel.getPlateColor()));
+        genericRowData.setField(10, StringData.fromString(travel.getCarType()));
+        genericRowData.setField(11, StringData.fromString(travel.getFirstLineId()));
+        genericRowData.setField(12, StringData.fromString(longToStandardFormatString(travel.getFirstTime())));
+        genericRowData.setField(13, travel.getFirstSpeed());
+        genericRowData.setField(14, StringData.fromString(travel.getArrivedLineId()));
+        genericRowData.setField(15, StringData.fromString(longToStandardFormatString(travel.getArrivedTime())));
+        genericRowData.setField(16, travel.getArrivedSpeed());
+        genericRowData.setField(17, StringData.fromString(travel.getInCrossLineId()));
+        genericRowData.setField(18, StringData.fromString(longToStandardFormatString(travel.getInCrossTime())));
+        genericRowData.setField(19, travel.getInSpeed());
+        genericRowData.setField(20, StringData.fromString(travel.getOutCrossLineId()));
+        genericRowData.setField(21, StringData.fromString(longToStandardFormatString(travel.getOutCrossTime())));
+        genericRowData.setField(22, travel.getOutSpeed());
+        genericRowData.setField(23, StringData.fromString(travel.getAwayLineId()));
+        genericRowData.setField(24, StringData.fromString(longToStandardFormatString(travel.getAwayTime())));
+        genericRowData.setField(25, travel.getAwaySpeed());
+        genericRowData.setField(26, StringData.fromString(travel.getLastLineId()));
+        genericRowData.setField(27, StringData.fromString(longToStandardFormatString(travel.getLastTime())));
+        genericRowData.setField(28, travel.getLastSpeed());
+        genericRowData.setField(29, travel.getCrossingTime());
+        genericRowData.setField(30, travel.getMinSpeed());
+        genericRowData.setField(31, travel.getMaxSpeed());
+        genericRowData.setField(32, travel.getAvgSpeed());
+        genericRowData.setField(33, Integer.valueOf(travel.getStopCount()));
+        genericRowData.setField(34, Integer.valueOf(travel.getStopTime()));
+        genericRowData.setField(35, StringData.fromString(longToStandardFormatString(System.currentTimeMillis())));
+        return genericRowData;
+    }
+}

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

@@ -4,12 +4,16 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
+import com.zhht.irn.functions.Travel2RowDataMapFunction;
 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;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,7 +42,21 @@ public class TravelInfoJob {
         // rawStringStream.sinkTo(FileUtils.getFileSink("./data/
 
         // 获取builder
-        DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_travel_metric");
+//        DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_travel_metric");
+
+        // 获取builder
+        String[] fields = { "record_id", "cross_id", "record_time", "travel_begin_time", "travel_end_time", "flow_direction", "car_id", "car_color", "plate", "plate_color",
+                "car_type", "first_lane_id", "first_time", "first_speed", "arrived_lane_id", "arrived_time", "arrived_speed", "incross_lane_id", "incross_time", "incross_speed",
+                "outcross_lane_id", "outcross_time", "outcross_speed", "away_lane_id", "away_time", "away_speed", "last_lane_id", "last_time", "last_speed", "duration",
+                "min_speed", "max_speed", "avg_speed", "stop_count", "stop_time", "statistic_time" };
+        DataType[] types = { DataTypes.VARCHAR(64), DataTypes.VARCHAR(64), DataTypes.DATE(), DataTypes.VARCHAR(32), DataTypes.VARCHAR(32), DataTypes.VARCHAR(16),
+                DataTypes.VARCHAR(64), DataTypes.VARCHAR(16), DataTypes.VARCHAR(32), DataTypes.VARCHAR(16), DataTypes.VARCHAR(16), DataTypes.VARCHAR(8),
+                DataTypes.VARCHAR(32), DataTypes.DOUBLE(), DataTypes.VARCHAR(8), DataTypes.VARCHAR(32), DataTypes.DOUBLE(), DataTypes.VARCHAR(8), DataTypes.VARCHAR(32),
+                DataTypes.DOUBLE(), DataTypes.VARCHAR(8), DataTypes.VARCHAR(32), DataTypes.DOUBLE(), DataTypes.VARCHAR(8), DataTypes.VARCHAR(32),
+                DataTypes.DOUBLE(), DataTypes.VARCHAR(8), DataTypes.VARCHAR(32), DataTypes.DOUBLE(), DataTypes.DOUBLE(), DataTypes.DOUBLE(), DataTypes.DOUBLE(), DataTypes.DOUBLE(),
+                DataTypes.INT(), DataTypes.INT(), DataTypes.VARCHAR(32) };
+
+        DorisSink.Builder<RowData> builder = DorisUtils.getBuilder("doris.properties", args, "app_travel_metric", fields, types);
 
         DataStream<Travel> travelStream = rawStringStream
                 // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
@@ -129,17 +147,9 @@ public class TravelInfoJob {
                     "停车总时长：{}", travel.getRecordId(), maxSpeed, minSpeed, travel.getAvgSpeed(), stopCount, stopTime);
 
             return travel;
-        }).map((MapFunction<Travel, String>) travel -> 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());
+        }).map(new Travel2RowDataMapFunction()).sinkTo(builder.build());
+
         env.execute("TravelInfoJob");
+
     }
 }

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

@@ -4,15 +4,17 @@ 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.RowDataSerializer;
 import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
 import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.sql.Connection;
 import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
+import java.util.Properties;
 
 public class DorisUtils {
 
@@ -33,83 +35,83 @@ public class DorisUtils {
         return DriverManager.getConnection(jdbcUrl, username, password);
     }
 
-    public static void close(Connection connection, PreparedStatement pstmt) {
-        if(null != pstmt) {
-            try {
-                pstmt.close();
-            } catch (SQLException e) {
-                e.printStackTrace();
-            }
-        }
-
-        if(null != connection) {
-            try {
-                connection.close();
-            } catch (SQLException e) {
-                e.printStackTrace();
-            }
-        }
-    }
-
     public static DorisSink.Builder<String> getBuilder(String filePath, String[] args, String table) throws IOException {
 
+        // 读取命令行参数，实例化argsTool
         ParameterTool argsTool = ParameterTool.fromArgs(args);
 
-        String label = argsTool.getRequired("label");
-
-        DorisSink.Builder<String> builder = DorisSink.builder();
-        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
-
+        // 读取配置文件，实例化tool
         InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream(filePath);
         ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
 
-        // 根据路径的配置文件获取MySQL链接参数
+        // 获取Doris相关参数
         String username = tool.getRequired("username");
         String password = tool.getRequired("password");
         String host = tool.getRequired("host");
         String port = tool.getRequired("http.port");
+        String label = argsTool.getRequired("label");
+
+        DorisSink.Builder<String> sinkBuilder = DorisSink.builder();
+        DorisOptions.Builder optionsBuilder = DorisOptions.builder();
+
         String database = tool.get("database", "default");
-        dorisBuilder.setFenodes(host + ":" + port)
+        optionsBuilder.setFenodes(host + ":" + port)
                 .setTableIdentifier(database + "." + table)
                 .setUsername(username)
                 .setPassword(password);
 
         DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
-        executionBuilder.setLabelPrefix(label); // streamload label prefix
+        executionBuilder.setLabelPrefix(label); // stream load label prefix
 
-        builder.setDorisReadOptions(DorisReadOptions.builder().build())
+        sinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
                 .setDorisExecutionOptions(executionBuilder.build())
                 .setSerializer(new SimpleStringSerializer()) //serialize according to string
-                .setDorisOptions(dorisBuilder.build());
-        return builder;
+                .setDorisOptions(optionsBuilder.build());
+        return sinkBuilder;
     }
 
-    public static DorisSink.Builder<String> getBuilder(String filePath,String label1) throws IOException {
-        DorisSink.Builder<String> builder = DorisSink.builder();
-        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
 
+    public static DorisSink.Builder<RowData> getBuilder(String filePath, String[] args, String table, String[] fields, DataType[] types) throws IOException {
+
+        // 读取命令行参数，实例化argsTool
+        ParameterTool argsTool = ParameterTool.fromArgs(args);
+
+        // 读取配置文件，实例化tool
         InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream(filePath);
         ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
 
-        // 根据路径的配置文件获取MySQL链接参数
+        // 获取Doris相关参数
         String username = tool.getRequired("username");
         String password = tool.getRequired("password");
         String host = tool.getRequired("host");
-        String port = tool.getRequired("port");
+        String port = tool.getRequired("http.port");
+        String label = argsTool.getRequired("label");
+
+        DorisSink.Builder<RowData> sinkBuilder = DorisSink.builder();
+        DorisOptions.Builder optionsBuilder = DorisOptions.builder();
+
         String database = tool.get("database", "default");
-        String table = tool.getRequired("table");
-        String label = tool.getRequired("label");
-        dorisBuilder.setFenodes(host + ":" + port)
+        optionsBuilder.setFenodes(host + ":" + port)
                 .setTableIdentifier(database + "." + table)
                 .setUsername(username)
                 .setPassword(password);
+
+        Properties properties = new Properties();
+        properties.setProperty("format", "json");
+        properties.setProperty("read_json_by_line", "true");
+
+
         DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
-        executionBuilder.setLabelPrefix(StringUtils.isNotEmpty(label1)?label1:label); // streamload label prefix
+        executionBuilder.setLabelPrefix(label); // stream load label prefix
+        executionBuilder.setStreamLoadProp(properties);
 
-        builder.setDorisReadOptions(DorisReadOptions.builder().build())
+        sinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
                 .setDorisExecutionOptions(executionBuilder.build())
-                .setSerializer(new SimpleStringSerializer()) //serialize according to string
-                .setDorisOptions(dorisBuilder.build());
-        return builder;
+                .setSerializer(RowDataSerializer.builder()
+                        .setFieldNames(fields)
+                        .setType("json")
+                        .setFieldType(types).build())//serialize according to string
+                .setDorisOptions(optionsBuilder.build());
+        return sinkBuilder;
     }
 }