lxb

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

@@ -50,7 +50,7 @@ object phaseGreenRatioJob {
     //查看获取到得topic数据
     //valueTopic.map(t => t.toString).print()
 
-    val dataStream: DataStreamSink[((String, String, String, String), Int)] = valueTopic.map(line => {
+    val dataStream: DataStream[Map] = valueTopic.map(line => {
       //启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
       //清场损失时间 = 黄灯时长 + 全红时长 - 2
       //绿信比 = 每个相位的绿灯时长累加 - 启动损失时间 - 清场损失时间
@@ -75,14 +75,15 @@ object phaseGreenRatioJob {
         allVolidGreen = allVolidGreen + green - startLossTime - clearLossTime;
       }
       var lxb = allVolidGreen/duration;
-    }).addSink(new MysqlSink);
+      val map1 = Map("id" -> 0, "lxb" -> lxb)
+      return map1
+    })
+    dataStream.addSink(new MysqlSink())
     //提交flink任务job
     environment.execute()
   }
-  
-  
-  
-  class MysqlSink extends RichSinkFunction[((String, String, String,String), Int)] {
+case class LXB(id: String, timestamp: Long, temp: Double)
+class MysqlSink extends RichSinkFunction[((String, String, String,String), Int)] {
     //获取配置文件
     val driver = ConfigUtil.getString("mysql-driver")
     val url = ConfigUtil.getString("mysql-url")