读取kafka数据计算后写入MySQL

realtime/hologram-streaming/pom.xml

@@ -13,6 +13,13 @@
     <version>1.0-SNAPSHOT</version>
 
     <dependencies>
+
+        <!-- flink 相关的依赖包 -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
@@ -28,6 +35,13 @@
             <artifactId>log4j-core</artifactId>
         </dependency>
 
+        <!-- 工具包 -->
+        <dependency>
+            <groupId>com.alibaba.fastjson2</groupId>
+            <artifactId>fastjson2</artifactId>
+            <version>2.0.12</version>
+        </dependency>
+
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java

+package com.zhht.irn.entity;
+
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 周期实体类
+ *
+ *
+ */
+public class Cycle {
+    // 周期编号
+    private int id;
+
+    // 路口编号
+    private String crossCode;
+
+    // 周期开始时间
+    private Date beginTime;
+
+    // 周期结束时间
+    private Date endTime;
+
+    private Long duration;
+
+    // 周期序号
+    private int cycleOrder;
+
+    // 周期内相位列表（依次执行）
+    private List<Phase> phaseList;
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getCrossCode() {
+        return crossCode;
+    }
+
+    public void setCrossCode(String crossCode) {
+        this.crossCode = crossCode;
+    }
+
+    public Date getBeginTime() {
+        return beginTime;
+    }
+
+    public void setBeginTime(Date beginTime) {
+        this.beginTime = beginTime;
+    }
+
+    public Date getEndTime() {
+        return endTime;
+    }
+
+    public void setEndTime(Date endTime) {
+        this.endTime = endTime;
+    }
+
+    public Long getDuration() {
+        return duration;
+    }
+
+    public void setDuration(Long duration) {
+        this.duration = duration;
+    }
+
+    public int getCycleOrder() {
+        return cycleOrder;
+    }
+
+    public void setCycleOrder(int cycleOrder) {
+        this.cycleOrder = cycleOrder;
+    }
+
+    public List<Phase> getPhaseList() {
+        return phaseList;
+    }
+
+    public void setPhaseList(List<Phase> phaseList) {
+        this.phaseList = phaseList;
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Phase.java

+package com.zhht.irn.entity;
+
+import java.util.Date;
+
+/**
+ * 相位实体类
+ */
+public class Phase {
+
+    // 相位编号
+    private String phaseValue;
+
+    // 相位开始时间
+    private Date beginTime;
+
+    // 相位结束时间
+    private Date endTime;
+
+    // 相位持续时长
+    private Long duration;
+
+    // 相位绿灯时间
+    private Long green;
+
+    // 相位黄灯时间
+    private Long yellow;
+
+    // 相位全红时间
+    private Long allRed;
+
+    public String getPhaseValue() {
+        return phaseValue;
+    }
+
+    public void setPhaseValue(String phaseValue) {
+        this.phaseValue = phaseValue;
+    }
+
+    public Date getBeginTime() {
+        return beginTime;
+    }
+
+    public void setBeginTime(Date beginTime) {
+        this.beginTime = beginTime;
+    }
+
+    public Date getEndTime() {
+        return endTime;
+    }
+
+    public void setEndTime(Date endTime) {
+        this.endTime = endTime;
+    }
+
+    public Long getDuration() {
+        return duration;
+    }
+
+    public void setDuration(Long duration) {
+        this.duration = duration;
+    }
+
+    public Long getGreen() {
+        return green;
+    }
+
+    public void setGreen(Long green) {
+        this.green = green;
+    }
+
+    public Long getYellow() {
+        return yellow;
+    }
+
+    public void setYellow(Long yellow) {
+        this.yellow = yellow;
+    }
+
+    public Long getAllRed() {
+        return allRed;
+    }
+
+    public void setAllRed(Long allRed) {
+        this.allRed = allRed;
+    }
+}

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

+package com.zhht.irn.flow;
+
+import com.alibaba.fastjson2.JSON;
+import com.zhht.irn.entity.Cycle;
+import com.zhht.irn.entity.Phase;
+import com.zhht.irn.sink.LXBSink;
+import com.zhht.irn.utils.FlinkUtils;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+
+import javax.xml.crypto.Data;
+import java.util.List;
+
+/**
+ * 两阶段提交参考文档：
+ * https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
+ */
+public class FlinkKafka2MysqlApp {
+    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<Tuple2<String, Double>> lxbStream = stream.map(new MapFunction<String, Tuple2<String, Double>>() {
+            @Override
+            public Tuple2<String, Double> map(String s) throws Exception {
+                //启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
+                //清场损失时间 = 黄灯时长 + 全红时长 - 2
+                //绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
+                Cycle cycle = JSON.parseObject(s, Cycle.class);
+                Long allValidGreen = 0L;
+                List<Phase> phaseList = cycle.getPhaseList();
+                for (Phase phase : phaseList) {
+                    // 绿灯时间
+                    long green = phase.getGreen();
+                    // 黄灯时间
+                    long yellow = phase.getYellow();
+                    // 全红时间
+                    long allRed = phase.getAllRed();
+
+                    // 启动损失时间取一般定义：2s
+                    long startLossTime = 2L;
+
+                    //清场损失时间
+                    long clearLossTime = yellow + allRed - 2;
+                    allValidGreen = allValidGreen + green - startLossTime - clearLossTime;
+                }
+
+                // 绿信比
+                Double lxb = allValidGreen.doubleValue() / cycle.getDuration().doubleValue();
+                return Tuple2.of(cycle.getCrossCode(), lxb);
+            }
+        });
+
+
+        lxbStream.addSink(new LXBSink());
+        env.execute();
+
+//        FlinkUtils.env.execute();
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/PKMySQLSink.java → realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LXBSink.java

@@ -2,6 +2,7 @@ package com.zhht.irn.sink;
 
 import com.zhht.irn.utils.MySQLUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 
@@ -11,7 +12,7 @@ import java.sql.PreparedStatement;
 /**
  * domain traffic
  */
-public class PKMySQLSink extends RichSinkFunction<Tuple2<String, Double>> {
+public class LXBSink extends RichSinkFunction<Tuple2<String, Double>> {
 
 
     Connection connection;
@@ -23,9 +24,11 @@ public class PKMySQLSink extends RichSinkFunction<Tuple2<String, Double>> {
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
 
-        connection = MySQLUtils.getConnection();
-        insertPstmt = connection.prepareStatement("insert into pk_traffic(domain,traffic) values (?,?)");
-        updatePstmt = connection.prepareStatement("update pk_traffic set traffic=? where domain=?");
+        // 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 wyf_test(cross_code,lxb) values (?,?)");
+        updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
 
     }
 

realtime/hologram-streaming/src/main/java/com/zhht/irn/source/StudentSource.java

@@ -15,7 +15,7 @@ public class StudentSource extends RichSourceFunction<Student> {
 
     @Override
     public void open(Configuration parameters) throws Exception {
-        connection = MySQLUtils.getConnection();
+        connection = MySQLUtils.getConnection(parameters.getString("path", ""));
         psmt = connection.prepareStatement("select * from student");
     }
 

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java

+package com.zhht.irn.utils;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.runtime.state.filesystem.FsStateBackend;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
+import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+
+public class FlinkUtils {
+
+    public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+
+    public static <T> DataStream<T> createKafkaStreamV3(String[] args, Class<? extends KafkaDeserializationSchema<T>> deser) throws Exception {
+        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
+
+        String groupId = tool.get("group.id", "test1");
+        String servers = tool.getRequired("bootstrap.servers");
+        List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
+        String autoCommit = tool.get("enable.auto.commit", "false");
+        String offsetReset = tool.get("auto.offset.reset", "earliest");
+
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", servers);
+        properties.setProperty("group.id", groupId);
+        properties.setProperty("enable.auto.commit", autoCommit);
+        properties.setProperty("auto.offset.reset",offsetReset);
+
+
+        int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
+        String checkpointPath = tool.get("checkpoint.path", "file:///Users/rocky/Desktop/Flink/workspace/imooc-flink/state");
+
+        env.enableCheckpointing(checkpointInterval);
+        env.setStateBackend(new FsStateBackend(checkpointPath));
+        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
+
+
+        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topics, deser.newInstance(), properties);
+
+        return env.addSource(kafkaConsumer);
+    }
+
+    public static <T> DataStream<T> createKafkaStreamV2(String[] args, Class<? extends DeserializationSchema<T>> deser) throws Exception {
+        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
+
+        String groupId = tool.get("group.id", "test1");
+        String servers = tool.getRequired("bootstrap.servers");
+        List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
+        String autoCommit = tool.get("enable.auto.commit", "false");
+        String offsetReset = tool.get("auto.offset.reset", "earliest");
+
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", servers);
+        properties.setProperty("group.id", groupId);
+        properties.setProperty("enable.auto.commit", autoCommit);
+        properties.setProperty("auto.offset.reset",offsetReset);
+
+
+        int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
+        String checkpointPath = tool.get("checkpoint.path", "file:///f/workspace/ZHHT-IRN-DB-ANALYSIS/state");
+
+        env.enableCheckpointing(checkpointInterval);
+        env.setStateBackend(new FsStateBackend(checkpointPath));
+        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
+
+
+        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topics, deser.newInstance(), properties);
+
+        return env.addSource(kafkaConsumer);
+    }
+
+    public static DataStream<String> createKafkaStreamV1(String[] args) throws Exception {
+        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
+
+        String groupId = tool.get("group.id", "test1");
+        String servers = tool.getRequired("bootstrap.servers");
+        List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
+        String autoCommit = tool.get("enable.auto.commit", "false");
+        String offsetReset = tool.get("auto.offset.reset", "earliest");
+
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", servers);
+        properties.setProperty("group.id", groupId);
+        properties.setProperty("enable.auto.commit", autoCommit);
+        properties.setProperty("auto.offset.reset",offsetReset);
+
+
+        int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
+        String checkpointPath = tool.get("checkpoint.path", "file:///Users/rocky/Desktop/Flink/workspace/imooc-flink/state");
+
+        env.enableCheckpointing(checkpointInterval);
+        env.setStateBackend(new FsStateBackend(checkpointPath));
+        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
+
+
+        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
+
+        return env.addSource(kafkaConsumer);
+    }
+
+
+    // /Users/rocky/Desktop/flink/11/pk.properties
+    public static void main(String[] args) throws Exception{
+        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
+
+        // 参数是分成2大类：1)必填 2)选填
+        String groupId = tool.get("group.id", "test1");
+        String servers = tool.getRequired("bootstrap.servers");
+
+        System.out.println(groupId);
+        System.out.println(servers);
+    }
+}

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/MySQLUtils.java

 package com.zhht.irn.utils;
 
+import org.apache.flink.api.java.utils.ParameterTool;
+
+import java.io.IOException;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
@@ -7,16 +10,20 @@ import java.sql.SQLException;
 
 public class MySQLUtils {
 
-    public static Connection getConnection() {
+    public static Connection getConnection(String propertyPath) throws Exception {
+        ParameterTool tool = ParameterTool.fromPropertiesFile(propertyPath);
 
-        try {
-            Class.forName("com.mysql.jdbc.Driver");
-            return DriverManager.getConnection("jdbc:mysql://127.0.0.1:13305/test", "root", "mima");
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
+        // 根据路径的配置文件获取MySQL链接参数
+        String driver = tool.getRequired("driver");
+        String username = tool.getRequired("username");
+        String password = tool.getRequired("password");
+        String host = tool.getRequired("host");
+        String port = tool.getRequired("port");
+        String database = tool.get("database", "default");
 
-        return null;
+        String jdbcUrl = String.format("jdbc:mysql://%s:%s/%s", host, port, database);
+        Class.forName(driver);
+        return DriverManager.getConnection(jdbcUrl, username, password);
     }
 
     public static void close(Connection connection, PreparedStatement pstmt) {
@@ -38,6 +45,10 @@ public class MySQLUtils {
     }
 
     public static void main(String[] args) {
-        System.out.println(getConnection());
+        try {
+            System.out.println(getConnection("F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties"));
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
     }
 }

realtime/hologram-streaming/src/main/resources/kafka.properties

+bootstrap.servers=124.71.213.187:9092
+kafka.input.topics=wyf-test-topic
\ No newline at end of file

realtime/hologram-streaming/src/main/resources/mysql.properties

+driver=com.mysql.jdbc.Driver
+host=127.0.0.1
+port=13305
+username=root
+password=mima
+database=test
\ No newline at end of file

realtime/pom.xml

@@ -24,6 +24,12 @@
 
             <dependency>
                 <groupId>org.apache.flink</groupId>
+                <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
+                <version>${flink.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.flink</groupId>
                 <artifactId>flink-clients_${scala.binary.version}</artifactId>
                 <version>${flink.version}</version>
             </dependency>