2022-12-19 临时提交中转

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

@@ -90,11 +90,11 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
         cycleSignalDataState = getRuntimeContext().getMapState(cycleSignalDataStateDescriptor);
         alreadyHandleCycleSignalDataState = getRuntimeContext().getMapState(alreadyHandleCycleSignalDataStateDescriptor);
         cycleSectionState = getRuntimeContext().getListState(cycleSectionStateDescriptor);
-        //  connection = DruidConnectPoolUtils.getConnection();
-     //   connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://localhost:3309/test?characterEncoding=utf8",
+         connection = DruidConnectPoolUtils.getConnection();
+/*     //   connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://localhost:3309/test?characterEncoding=utf8",
         connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
 
-                "root","mima").getConnection();
+                "root","mima").getConnection();*/
 
         dim_cnt_cross_lane_position = getLaneDimData(connection);
 
@@ -174,6 +174,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             cycleSectionState.add(value.getBeginDateTime());
 
             String crossCode = value.getCrossCode();
+          //  System.out.println("当前路口是---->"+crossCode);
             Integer cycleOrder = value.getCycleOrder();
             //周期开始时间，周期结束时间
             Long beginDateTime = value.getBeginDateTime();
@@ -473,7 +474,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
      */
     private Map<String, Map<String, String>> getLaneDimData(DruidPooledConnection connection) throws SQLException {
 
-        String sql = "select cross_id, lane_id,position,concat(position , name ) as flow_direction from test.dim_cnt_cross_lane_position  ";
+        String sql = "select cross_id, lane_id,position,turn_direction as flow_direction from dim_cnt_cross_lane_position  ";
         PreparedStatement ps = connection.prepareStatement(sql);
         ResultSet resultSet = ps.executeQuery();
 
@@ -765,7 +766,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
 
 
-            String sql = "INSERT INTO test.app_cross_line_travel_car_info" +
+            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, " +

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

@@ -90,7 +90,7 @@ public class TravelSituationAnalysisJob {
             // FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
             //   FlinkKafkaConsumer<String> travelInfo = new FlinkKafkaConsumer<>("t_info2", new SimpleStringSchema(), kafkaProperties);
             FlinkKafkaConsumer<TravelInfo> travelInfo = new FlinkKafkaConsumer<>("trips_info", new TravelInfoKafkaSchema(), kafkaProperties);
-            travelInfo.setStartFromEarliest();
+          //  travelInfo.setStartFromEarliest();
             DataStreamSource<TravelInfo> travelInfoStream = env.addSource(travelInfo).setParallelism(3);
 
             //  travelInfoStream.print();
@@ -98,7 +98,7 @@ public class TravelSituationAnalysisJob {
             //周期信号
             // FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
             FlinkKafkaConsumer<CycleSignalData> cycleSignalData = new FlinkKafkaConsumer<>("signal_cycle_data", new CycleSignalKafkaSchema(), kafkaProperties2);
-            cycleSignalData.setStartFromEarliest();
+           // cycleSignalData.setStartFromEarliest();
             DataStreamSource<CycleSignalData> cycleSignalDataStream = env.addSource(cycleSignalData);
 
             // cycleSignalDataStream.print();

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

@@ -26,8 +26,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
-                "root","mima").getConnection();
+        connection = DruidConnectPoolUtils.getConnection();
 
     }
 
@@ -41,7 +40,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
     public void invoke(List<TravelLineSinkInfo> value, Context context) throws Exception {
         super.invoke(value, context);
 
-        String sql = "INSERT INTO test.app_cross_line_travel_car_info" +
+        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, " +