绿信比

realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/GreenRadioJob.java

+package com.zhht.irn.flow;
+
+import com.alibaba.fastjson.JSON;
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.Phase;
+import com.zhht.irn.sink.GreenRadioSink;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import java.util.List;
+
+/**
+ * 两阶段提交参考文档：
+ * https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
+ */
+public class GreenRadioJob {
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+//        DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
+        // 目前假设从Kafka接入的数据为json格式的字符串
+        // Cycle Topic 的消息格式如下：
+        // {"Id":"","CrossCode":"","BeginTime":"","EndTime":"","Duration":"","CycleOrder":"",
+        //         "detail":[{"PhaseValue":"101","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
+        //                   {"PhaseValue":"102","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
+        //                   {"PhaseValue":"103","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
+        //                   {"PhaseValue":"104","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}]
+
+        // 造一条数据
+        DataStream<String> stream = env.fromElements("{\"id\":\"1\",\"crossCode\":\"0001\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"phaseList\":[{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
+        DataStream<Cycle> lxbStream = stream.map(new MapFunction<String, Cycle>() {
+            @Override
+            public Cycle map(String s) throws Exception {
+                //启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
+                //清场损失时间 = 黄灯时长 + 全红时长 - 2
+                //绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
+                Cycle cycle = JSON.parseObject(s, Cycle.class);
+                long cycleId= cycle.getId();
+                long duration = cycle.getDuration();
+                String crossCode = cycle.getCrossCode();
+                double cycleGreenRadio = 0;
+                List<Phase> phaseList = cycle.getPhaseList();
+                for (Phase phase : phaseList) {
+                    //相位号
+                    String phaseValue = phase.getPhaseValue();
+                    // 绿灯时间
+                    long green = phase.getGreen();
+                    // 黄灯时间
+                    long yellow = phase.getYellow();
+                    // 全红时间
+                    long allRed = phase.getAllRed();
+                    // 启动损失时间取一般定义：2s
+                    long startLossTime = 2L;
+                    //清场损失时间
+                    long clearLossTime = yellow + allRed - 2;
+                    //阶段有效绿灯时间
+                    long phaseValidGreen = green + yellow + allRed - startLossTime - clearLossTime;
+                    //阶段绿信比
+                    double phaseGreenRadio = phaseValidGreen/duration;
+                    //组装阶段绿信比
+                    phase.setGreenRatio(phaseGreenRadio);
+                    phase.setValidGreen(phaseValidGreen);
+                    //周期绿信比= 各个阶段绿信比之和
+                    cycleGreenRadio = cycleGreenRadio + phaseGreenRadio;
+                }
+                // 组装周期绿信比
+                cycle.setGreenRadio(cycleGreenRadio);
+                return cycle;
+            }
+        });
+        lxbStream.addSink(new GreenRadioSink());
+        env.execute();
+    }
+}
\ No newline at end of file

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

+package com.zhht.irn.sink;
+
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.utils.MySQLUtils;
+import org.apache.flink.api.java.tuple.Tuple2;
+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 GreenRadioSink extends RichSinkFunction<Cycle> {
+
+
+    Connection connection;
+    PreparedStatement insertPstmt;
+    PreparedStatement updatePstmt;
+
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+
+        // path暂定为固定值
+        String path = "F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
+        connection = MySQLUtils.getConnection(path);
+        insertPstmt = connection.prepareStatement("insert into app_cycle_green_ratio(id,cross_code,green_radio) values (?,?,?)");
+        //updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
+    }
+
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if(insertPstmt != null) insertPstmt.close();
+        if(updatePstmt != null) updatePstmt.close();
+        if(connection != null) connection.close();
+    }
+
+    /**
+     * 来一条数据就执行一次
+     *
+     * 1000w的数据  1000w次
+     */
+    @Override
+    public void invoke(Cycle value, Context context) throws Exception {
+        System.out.println("=====invoke======" + value.getId() + "-->" + value.getCrossCode());
+        insertPstmt.setInt(1, value.getId());
+        insertPstmt.setString(2, value.getCrossCode());
+        insertPstmt.setDouble(3,value.getGreenRadio());
+        insertPstmt.execute();
+    }
+}