行车情况提交最新代码，读取配置文件方式

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

@@ -5,6 +5,7 @@ import com.alibaba.druid.pool.DruidPooledConnection;
 
 
 import com.zhht.irn.entity.dto.*;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.DruidConnectPoolUtils;
 import org.apache.flink.api.common.state.*;
 import org.apache.flink.api.common.time.Time;
@@ -14,6 +15,7 @@ import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
 import org.apache.flink.util.Collector;
 import org.apache.log4j.Logger;
 
+import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -45,6 +47,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
+    Connection connection;
+
 
     @Override
     public void open(Configuration parameters) throws Exception {
@@ -84,11 +88,18 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
         dim_cnt_cross_lane_position = getLaneDimData();
 
+
+         connection = DorisUtils.getConnection();
+
+
     }
 
     @Override
     public void close() throws Exception {
         super.close();
+        if(connection!=null){
+            connection.close();
+        }
     }
 
     //把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
@@ -223,7 +234,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             long currentProcessingTime = ctx.timerService().currentProcessingTime();
             // 只在某个时间段注册定时器
             SimpleDateFormat sdf = new SimpleDateFormat("HH");
-            if (sdf.format(currentProcessingTime).equals("11")) {
+            if (sdf.format(currentProcessingTime).equals("14")) {
                 ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 30 * 60 * 1000);
             }
 
@@ -467,12 +478,12 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
      * @return
      * @throws SQLException
      */
-    private Map<String, Map<String, String>> getLaneDimData() throws SQLException {
+    private Map<String, Map<String, String>> getLaneDimData() throws Exception {
 
 
-        DruidPooledConnection connection1 = DruidConnectPoolUtils.getConnection();
+        connection=DorisUtils.getConnection();
         String sql = "select cross_id, lane_id,position,turn_direction as flow_direction from dim_cnt_cross_lane_position  ";
-        PreparedStatement ps = connection1.prepareStatement(sql);
+        PreparedStatement ps = connection.prepareStatement(sql);
         ResultSet resultSet = ps.executeQuery();
 
         Map<String, Map<String, String>> dimData = new HashMap<>();
@@ -491,8 +502,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             }
         }
 
-        if(connection1!=null){
-            connection1.close();
+        if(connection!=null){
+            connection.close();
         }
 
         return dimData;

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

@@ -10,10 +10,12 @@ 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.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.ConnectedStreams;
@@ -25,6 +27,7 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
 
+import java.io.InputStream;
 import java.util.List;
 import java.util.Properties;
 
@@ -39,22 +42,27 @@ public class TravelSituationAnalysisJob {
 
         try {
 
+            InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("kafka.properties");
+            ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
+
+
 //1、创建环境
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             // checkpoint配置
             CheckpointConfig checkpointConfig = env.getCheckpointConfig();
             // checkpoint时间间隔3分钟
-            checkpointConfig.setCheckpointInterval(3 * 60 * 1000);
+            checkpointConfig.setCheckpointInterval( 60 * 1000);
 
             // 两次checkpoint中最短时间间隔1分钟
             checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
             // 同时同时允许1个checkpoint进行
             checkpointConfig.setMaxConcurrentCheckpoints(1);
 
-            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
-
             //这如果不指定使用FsStateBackend 则默认使用flink集群中的配置，rocksDB
-            env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
+         //   env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
+            env.setStateBackend(new FsStateBackend(tool.getRequired("checkpoint.path")));
+            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+            env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
             //设置重启策略 这里不设置重启策略，由平台统一拉起任务
             // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
@@ -63,8 +71,9 @@ public class TravelSituationAnalysisJob {
 
 
             Properties kafkaProperties = new Properties();
-             kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
-          //  kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
+           //  kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
+             kafkaProperties.setProperty("bootstrap.servers", tool.getRequired("bootstrap.servers"));
+          // kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
             //  kafkaProperties.setProperty("group.id", "travel_event");
             kafkaProperties.setProperty("group.id", "trips_info2");
             kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
@@ -72,8 +81,9 @@ public class TravelSituationAnalysisJob {
 
 
             Properties kafkaProperties2 = new Properties();
-              kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
-          //  kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
+          //  kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
+            kafkaProperties2.setProperty("bootstrap.servers", tool.getRequired("bootstrap.servers"));
+           // kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
             kafkaProperties2.setProperty("group.id", "cycle_signal2");
             kafkaProperties2.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
             kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

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

@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
 import com.alibaba.druid.pool.DruidPooledConnection;
 import com.zhht.irn.entity.dto.TravelLineSinkInfo;
 import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
+import com.zhht.irn.utils.DorisUtils;
 import com.zhht.irn.utils.DruidConnectPoolUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
@@ -31,7 +32,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = DruidConnectPoolUtils.getConnection();
+        connection = DorisUtils.getConnection();
         logger.info("建立数据库连接成功。。。。。。。");
 
     }
@@ -43,6 +44,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
             connection.close();
         }
 
+
     }
 
     @Override
@@ -50,7 +52,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
         super.invoke(value, context);
 
         //重新获取连接，避免超时的问题
-        connection = DruidConnectPoolUtils.getConnection();
+        connection = DorisUtils.getConnection();
         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," +
@@ -93,6 +95,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
         int[] ints = preparedStatement.executeBatch();
 
         logger.info("写入数据成功。。。。。"+ints.length+"...");
+        preparedStatement.close();
 
         connection.close();