旅行信息

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

@@ -4,7 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.entity.mysql.TripsMetric;
-import com.zhht.irn.sink.TripsMetricSink;
+import com.zhht.irn.sink.TravelMetricSink;
 import org.apache.flink.api.common.functions.FilterFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
@@ -18,7 +18,6 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
@@ -71,7 +70,6 @@ public class TravelInfoJob {
         // 基于Kafka 原始JSON格式的字符串，获取消息流
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
         DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
-
         // 开始计算旅行信息相关指标
         rawStringStream.map(new MapFunction<String, Travel>() {
             @Override
@@ -120,22 +118,7 @@ public class TravelInfoJob {
                 travel.setMinSpeed(minSpeed);
                 return travel;
             }
-        }).filter(new FilterFunction<Travel>() {
-            @Override
-            public boolean filter(Travel travelInfo) throws Exception {
-                // 过滤掉垃圾数据
-                return true;
-            }
-        }).map(new MapFunction<Travel, TripsMetric>() {
-            @Override
-            public TripsMetric map(Travel travelInfo) throws Exception {
-                TripsMetric tripsMetric = new TripsMetric();
-                // 计算各类指标, 塞入tripsMetric @todo
-                List<Location> locations = travelInfo.getLocations();
-
-                return tripsMetric;
-            }
-        }).addSink(new TripsMetricSink()); //@todo 插入数据库的逻辑
+        }).addSink(new TravelMetricSink()); //@todo 插入数据库的逻辑
 
     }
 

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

+package com.zhht.irn.sink;
+
+import com.zhht.irn.entity.Travel;
+import com.zhht.irn.entity.mysql.TripsMetric;
+import com.zhht.irn.utils.DateUtils;
+import com.zhht.irn.utils.MySQLUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+
+/**
+ * domain traffic
+ */
+public class TravelMetricSink extends RichSinkFunction<Travel> {
+    Connection connection;
+    PreparedStatement insertPstmt;
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        // path暂定为固定值
+        String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
+        connection = MySQLUtils.getConnection(path);
+        String insertSql="insert into app_travel_metric(record_id,cross_id,car_id,plate,capture_time,lost_time,stop_count,stop_time,ave_speed,min_speed,max_speed,update_time) values (?,?,?,?,?,?,?,?,?,?,?,?)";
+        insertPstmt = connection.prepareStatement(insertSql);
+    }
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if(insertPstmt != null) insertPstmt.close();
+        if(connection != null) connection.close();
+    }
+
+    /**
+     * 来一条数据就执行一次
+     *
+     * 1000w的数据  1000w次
+     */
+    @Override
+    public void invoke(Travel travel, Context context) throws Exception {
+        System.out.println("=====invoke======" + travel.getId() + "-->" + travel.getRecordId());
+        insertPstmt.setString(1, travel.getRecordId());
+        insertPstmt.setString(2, travel.getCrossId());
+        insertPstmt.setLong(3, travel.getCarId());
+        insertPstmt.setString(4,travel.getPlate());
+        insertPstmt.setString(5,travel.getFirstTime());
+        insertPstmt.setString(6,travel.getLastTime());
+        insertPstmt.setInt(7,travel.getStopCount());
+        insertPstmt.setDouble(8,travel.getStopTime());
+        insertPstmt.setDouble(9,travel.getAveSpeed());
+        insertPstmt.setDouble(10,travel.getMinSpeed());
+        insertPstmt.setDouble(11,travel.getMaxSpeed());
+        insertPstmt.setString(12, DateUtils.getTodayTime());
+        insertPstmt.execute();
+    }
+}

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

-package com.zhht.irn.sink;
-
-import com.zhht.irn.entity.Cycle;
-import com.zhht.irn.entity.Stage;
-import com.zhht.irn.entity.mysql.TripsMetric;
-import com.zhht.irn.utils.MySQLUtils;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.util.List;
-
-/**
- * domain traffic
- */
-public class TripsMetricSink extends RichSinkFunction<TripsMetric> {
-    Connection connection;
-    PreparedStatement insertCyclePstmt;
-    PreparedStatement insertPhasePstmt;
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        // path暂定为固定值
-//        String path = "F:\\workbench\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
-//        connection = MySQLUtils.getConnection(path);
-//        String cycleSql="insert into app_cycle_green_ratio(cycle_id,cross_code,begin_time,end_time,duration,green_ratio) values (?,?,?,?,?,?)";
-//        insertCyclePstmt = connection.prepareStatement(cycleSql);
-//        String phaseSql="insert into app_phase_green_ratio(cycle_id,cross_code,phase_value,begin_time,end_time,duration,green_ratio,valid_green) values (?,?,?,?,?,?,?,?)";
-//        insertPhasePstmt = connection.prepareStatement(phaseSql);
-    }
-    @Override
-    public void close() throws Exception {
-        super.close();
-        if(insertCyclePstmt != null) insertCyclePstmt.close();
-        if(insertPhasePstmt != null) insertPhasePstmt.close();
-        if(connection != null) connection.close();
-    }
-
-    /**
-     * 来一条数据就执行一次
-     *
-     * 1000w的数据  1000w次
-     */
-    @Override
-    public void invoke(TripsMetric tripsMetric, Context context) throws Exception {
-//        System.out.println("=====invoke======" + cycle.getId() + "-->" + cycle.getCrossCode());
-//        insertCyclePstmt.setLong(1, cycle.getId());
-//        insertCyclePstmt.setString(2, cycle.getCrossCode());
-//        insertCyclePstmt.setString(3, cycle.getBeginTime());
-//        insertCyclePstmt.setString(4, cycle.getEndTime());
-//        insertCyclePstmt.setInt(5,cycle.getDuration());
-//        insertCyclePstmt.setDouble(6,cycle.getGreenRatio());
-//        insertCyclePstmt.execute();
-//        List<Phase> phaseList = cycle.getPhaseList();
-//        for(int i=0;i<phaseList.size();i++){
-//            Phase phase = phaseList.get(i);
-//            insertPhasePstmt.setLong(1,cycle.getId());
-//            insertPhasePstmt.setString(2,phase.getCrossCode());
-//            insertPhasePstmt.setString(3,phase.getPhaseValue());
-//            insertPhasePstmt.setString(4, cycle.getBeginTime());
-//            insertPhasePstmt.setString(5, cycle.getEndTime());
-//            insertPhasePstmt.setInt(6,cycle.getDuration());
-//            insertPhasePstmt.setDouble(7,phase.getGreenRatio());
-//            insertPhasePstmt.setInt(8,phase.getValidGreen());
-//            insertPhasePstmt.addBatch();
-//        }
-//        insertPhasePstmt.executeBatch();
-//        insertPhasePstmt.clearBatch();
-    }
-}