增加fast

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

@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.Stage;
 import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.MySQLUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
@@ -28,10 +29,10 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         // path暂定为固定值
-        String path = ".\\realtime\\hologram-streaming\\src\\main\\resources\\doris.properties";
-        connection = MySQLUtils.getConnection(path);
+        //String path = ".\\realtime\\hologram-streaming\\src\\main\\resources\\doris.properties";
+        connection = DorisUtils.getConnection();
         String cycleSql="insert into app_cycle_green_ratio" +
-                "(id,cross_id,signal_code,begin_time,end_time,duration,cycle_order,green_ratio,loss_time,statistic_time,record_date) " +
+                "(id,cycle_order,cross_id,signal_code,begin_time,end_time,duration,green_ratio,loss_time,statistic_time,record_date) " +
                 "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         insertCyclePstmt = connection.prepareStatement(cycleSql);
 
@@ -48,17 +49,16 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
     @Override
     public void invoke(Cycle cycle, Context context) throws Exception {
         insertCyclePstmt.setLong(1, cycle.getId());
-        insertCyclePstmt.setString(2, cycle.getCrossCode());
-        insertCyclePstmt.setString(3, cycle.getSignalCode());
-        insertCyclePstmt.setString(4, cycle.getBeginDateTime());
-        insertCyclePstmt.setString(5, cycle.getEndDateTime());
-        insertCyclePstmt.setInt(6, cycle.getDuration());
-        insertCyclePstmt.setLong(7, cycle.getCycleOrder());
+        insertCyclePstmt.setLong(2, cycle.getCycleOrder());
+        insertCyclePstmt.setString(3, cycle.getCrossCode());
+        insertCyclePstmt.setString(4, cycle.getSignalCode());
+        insertCyclePstmt.setString(5, DateUtils.stampToTime(cycle.getBeginDateTime()));
+        insertCyclePstmt.setString(6, DateUtils.stampToTime(cycle.getEndDateTime()));
+        insertCyclePstmt.setInt(7, cycle.getDuration());
         insertCyclePstmt.setDouble(8, cycle.getGreenRatio());
         insertCyclePstmt.setInt(9, cycle.getLossTime());
         insertCyclePstmt.setString(10, DateUtils.getTodayTime());
         insertCyclePstmt.setString(11, DateUtils.getTodayDate());
-        //insertCyclePstmt.setString(12, DateUtils.getCurrentMonth());
         insertCyclePstmt.execute();
 
         //List<Stage> stageList = cycle.getStageList();