绿信比

realtime/hologram-streaming/src/main/scala/com/zhht/irn/PhaseGreenRatioJob.scala

+package com.zhht.irn
+import java.sql.{Connection, DriverManager, PreparedStatement}
+import java.text.SimpleDateFormat
+import org.apache.flink.api.common.restartstrategy.RestartStrategies
+import org.apache.flink.streaming.api.CheckpointingMode
+import org.apache.flink.streaming.api.environment.{CheckpointConfig, StreamExecutionEnvironment}
+import org.apache.flink.streaming.api.scala.DataStream
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import com.alibaba.fastjson.JSON
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink}
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.streaming.api.windowing.time.Time
+
+import java.util.Properties
+import com.jt.util.ConfigUtil
+import org.apache.flink.api.common.serialization.SimpleStringSchema
+import org.apache.flink.streaming.api.scala._
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
+
+
+object phaseGreenRatioJob {
+  def main(args: Array[String]): Unit = {
+    //创建流处理环境
+    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
+    //之后开启CheckPointing可以开启重启策略
+    environment.enableCheckpointing(5000)
+    //设置重启策略为，出现三次异常重启3次,隔10秒一次
+    environment.getConfig.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000))
+    //系统异常退出或者人为退出，不删除checkpoint数据
+    environment.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
+    //设置Checkpoint模式(与Kafka整合,要设置Checkpoint模式为Exactly_Once)
+    environment.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
+    environment.setParallelism(1)
+    //environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+    //配置kafka信息
+    val properties = new Properties()
+    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "srv191:9092,srv192:9092,srv193:9092")
+    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "console-consumer-3145")
+    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
+    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
+    //如果没有记录偏移量，第一次从最开始消费:earliest 从最新的位置开始消费:latest
+    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
+    //kafka的消费者，不自动提交偏移量
+    properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+
+    // 获取topic数据
+    val valueTopic: DataStream[String] = environment.addSource(new FlinkKafkaConsumer[String]("kafka", new SimpleStringSchema(), properties))
+    //查看获取到得topic数据
+    //valueTopic.map(t => t.toString).print()
+
+    val dataStream: DataStreamSink[((String, String, String, String), Int)] = valueTopic.map(line => {
+      //启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
+      //清场损失时间 = 黄灯时长 + 全红时长 - 2
+      //绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
+      //消息格式：{"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":""}]
+      val Phases = JSON.parseObject(line).getString("detail");
+      var duration = JSON.parseObject(line).getString("duration");
+      //有效绿灯时间
+      var allVolidGreen = 0;
+      for(Phase <- Phases){
+        var green = Phase.getString("Green").toLong;
+        var yellow = Phase.getString("Yellow").toLong;
+        var allRed = Phase.getString("AllRed").toLong;
+
+        //启动损失时间
+        var startLossTime = 2;
+        //清场损失时间
+        var clearLossTime = yellow + allRed - 2;
+        allVolidGreen = allVolidGreen + green - startLossTime - clearLossTime;
+      }
+      var lxb = allVolidGreen/duration;
+    }).addSink(new MysqlSink);
+    //提交flink任务job
+    environment.execute()
+  }
+  
+  
+  
+  class MysqlSink extends RichSinkFunction[((String, String, String,String), Int)] {
+    //获取配置文件
+    val driver = ConfigUtil.getString("mysql-driver")
+    val url = ConfigUtil.getString("mysql-url")
+    val user = ConfigUtil.getString("mysql-user")
+    val password = ConfigUtil.getString("mysql-password")
+
+    private var connection: Connection = null
+    private var ps: PreparedStatement = null
+
+    override def open(parameters: Configuration): Unit = {
+      //1:加载驱动
+      Class.forName(driver)
+      //2：创建连接
+      connection = DriverManager.getConnection(url, user, password)
+      //3:获得执行语句
+      val sql = "insert into person_count(db,TableName,TypeTable,timeStamp,person_count) values(?,?,?,?,?);"
+      ps = connection.prepareStatement(sql)
+    }
+
+    override def invoke(value: ((String, String, String,String), Int)): Unit = {
+      try {
+        //4.组装数据，执行插入操作
+        ps.setString(1, value._1._1)
+        ps.setString(2, value._1._2)
+        ps.setString(3, value._1._3)
+        ps.setString(4, value._1._4)
+        ps.setInt(5, value._2)
+        ps.executeUpdate()
+      } catch {
+        case e: Exception => println(e.getMessage)
+      }
+    }
+
+    //关闭连接操作
+    override def close(): Unit = {
+      if (connection != null) {
+        connection.close()
+      }
+      if (ps != null) {
+        ps.close()
+      }
+    }
+  }
+}