行车情况sink doris 使用connector

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

+package com.zhht.irn.functions;
+
+import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.util.Collector;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * TravelLineSinkInfoFlatMapFunction
+ *
+ * @author Rui meng
+ * @description 行车情况切分为字符串
+ * @date 2023/1/9 16:42
+ */
+public class TravelLineSinkInfoFlatMapFunction implements FlatMapFunction<List<TravelLineSinkInfo>, String> {
+
+    @Override
+    public void flatMap(List<TravelLineSinkInfo> travelLineSinkInfos, Collector<String> collector) throws Exception {
+
+        //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:travelLineSinkInfos) {
+            // 使用事件数据的时间划分数据分区，这样可以进行补数操作
+           String record_date = sdf2.format(t.getLast_car_inCross_time());
+          //  String record_date = "2023-01-09";
+
+           collector.collect(record_date+"\t"+t.getCross_id()+"\t"+statistic_time+"\t"+t.getCycle_begin_time()+"\t"+t.getCycle_end_time()
+           +"\t"+t.getLane_id()+"\t"+t.getCycle_id()+"\t"+t.getDirection()+"\t"+t.getFlow_direction()+"\t"+t.getPass_numbers()
+           +"\t"+t.getLast_car_inCross_time()+"\t"+t.getAverage_pass_time()+"\t"+t.getAverage_pass_speed()+"\t"+t.getAverage_control_delay()
+           +"\t"+t.getAverage_stop_delay()+"\t"+t.getAverage_stop_times()+"\t"+ t.getAverage_car_head_time_gap());
+
+        }
+    }
+}

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

@@ -6,11 +6,14 @@ 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;
@@ -24,6 +27,7 @@ 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;
 
 
@@ -65,7 +69,7 @@ public class TravelSituationAnalysisJob {
             env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
             //设置重启策略 这里不设置重启策略，由平台统一拉起任务
-            // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
+            //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
             env.setRestartStrategy(RestartStrategies.noRestart());
 
 
@@ -90,16 +94,16 @@ public class TravelSituationAnalysisJob {
 
 
             //旅行信息数据
-         //   FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
-            FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
+          // FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("t_info2", new TravelInfoKafkaSchema(), kafkaProperties);
+          FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
             DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
+
             //周期信号
-        //    FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
-            FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
+         //   FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("s_data2", new CycleSignalKafkaSchema(), kafkaProperties2);
+           FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
             DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
 
-
             //2、使用connect，把2个流联合处理
             ConnectedStreams<TravelInfo, CycleSignalData> connect = travelInfoStream
                     .connect(cycleSignalDataStream);
@@ -112,7 +116,13 @@ public class TravelSituationAnalysisJob {
             SingleOutputStreamOperator<List<TravelLineSinkInfo>> sinkDataStream = travelInfoCycleSignalDataConnectedStreams
                     .process(new TravelCarInfoCoProcessFunction()).setParallelism(3);
 
-            sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
+           // sinkDataStream.addSink(new TravelLaneInfoSink()).setParallelism(3);
+
+
+            // 获取builder
+            DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-travel-situation.properties");
+
+            sinkDataStream.flatMap(new TravelLineSinkInfoFlatMapFunction()).sinkTo(builder.build());
 
             env.execute("TravelSituationAnalysisJob");
 

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