任务概述

  • 对应指标：车辆实时位置、车辆实时下一站、车辆实时舒适度、车辆实时状态、实时车速、实时累计行驶里程、实时车型状态分布
  • 开发语言Scala，实时计算引擎flink

数据流向

image.png

  1. 实时数据流
    • 公交车定时(10s)通过http发送can总线原包数据到tcp服务
    • tcp服务对http原始数据包进行解包，并按指定分隔符(FAF5)切割数据，发送(字节数组)到kafka
    • flink利用自定义反序列化方法将字节数组转化成字符串
    • 对gps数据进行拆分截取，并根据深标协议转码
    • 根据相关ID到hbase取得维度信息
    • 进一步计算相关指标，并将数据sink到后端mysql
  2. 维表
    • 各业务库原始数据由平台采集同步到平台ods层
    • 利用hql进行清洗转化得到dim层维度数据
    • 用dataX将维表数据同步到hbse，供flink程序进行join

code

source 类

  • 利用flink自带的kafka connector FlinkKafkaConsumer09创建DataSource
  • 由于kafka中发送的数据为字节数组，所以序列化方法改用自定义的序列化方法 ByteArrayDeserializationSchema ```scala package com.szbus.source

import java.util import java.util.Properties import com.szbus.util.ByteArrayDeserializationSchema import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.api.scala._ import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition

/*

  • @author : 恋晨
  • Date : 2019/7/12 3:44 PM
  • 功能 : 获取各个数据流的source / class source {

/**

  1. * 设置kafka参数
  2. */

val properties = new Properties() properties.setProperty(“bootstrap.servers”, “10.128.1.13:9092,10.128.1.18:9092,10.128.1.19:9092”) properties.setProperty(“group.id”, “prd”) / key 反序列化 */ properties.setProperty(“key.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”) / value 反序列化 */ properties.setProperty(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”)

/**

  1. *               获取gps数据
  2. * @param env    flink运行环境
  3. * @return       返回gps source
  4. */

def getGpsSource(env:StreamExecutionEnvironment,parallelism:Int): DataStream[String] ={

  1.  val gpsSource = env
  2.           .addSource(new FlinkKafkaConsumer09[String](
  3.             "bus_gps",
  4.             new ByteArrayDeserializationSchema(),
  5.             properties
  6.           ).setStartFromLatest()).name("gpsSource")
  7.          .setParallelism(parallelism)
  8.  gpsSource

} }

  2. 自定义序列化方法
  3. ```java
  4. package com.szbus.util;
  6. import org.apache.flink.api.common.serialization.DeserializationSchema;
  7. import org.apache.flink.api.common.serialization.SerializationSchema;
  8. import org.apache.flink.api.common.typeinfo.TypeInformation;
  10. import java.io.IOException;
  12. /**
  13.  * @author : 恋晨
  14.  * Date  : 2019/6/28 5:57 PM
  15.  * 功能  : kafka Topic 字节数组反序列化类
  16.  */
  17. public class ByteArrayDeserializationSchema implements DeserializationSchema<String>,SerializationSchema<String>{
  19.     public static  String parseByte2HexStr(byte buf[]){
  20.         StringBuffer sb = new StringBuffer();
  21.         for (int i = 0 ; i < buf.length ; i++){
  22.             String hex = Integer.toHexString(buf[i] & 0xFF);
  23.             if(hex.length() == 1){
  24.                 hex = '0' + hex;
  25.             }
  26.             sb.append(hex.toUpperCase());
  27.         }
  29.         return sb.toString();
  30.     }
  32.     @Override
  33.     public String deserialize(byte[] bytes) throws IOException {
  34.         String out = parseByte2HexStr(bytes);
  35.         return out;
  36.     }
  38.     @Override
  39.     public boolean isEndOfStream(String s) {
  40.         return false;
  41.     }
  43.     @Override
  44.     public byte[] serialize(String s) {
  45.         return new byte[0];
  46.     }
  49.     @Override
  50.     public TypeInformation<String> getProducedType() {
  51.         return TypeInformation.of(String.class);
  52.     }
  53. }

join 类

  • 这个类继承AsyncFunction，利用flink异步IO进行流维join
  • 输入输出数据类型，分别数case class{ gps , gps_join_stop_previous}
  • 从输入数据中取出bus_id和line_code，用这两个rowKey从维表dim_hbase_line_stop，dim_hbase_line_stop取得对应的数据(json格式)
  • 从json取出需要的字段(up_end_order:上行终点站序号,down_end_order:下行终点站序号,stop_list:站点列表,bus_type:车辆型号)，组合成新的case class gps_join_stop_previous
  1. package com.szbus.join
  2. import com.szbus.cache.HbaseSideCache
  3. import org.apache.flink.streaming.api.scala.async.{AsyncFunction, ResultFuture}
  4. import com.szbus.main.caseClasses.gps
  5. import com.szbus.main.caseClasses.gps_join_stop_previous
  6. /**
  7.   *
  8.   * @author : 恋晨
  9.   *         Date  : 2019/7/23 2:26 PM
  10.   *         功能  :
  11.   *
  12.   */
  13. class gpsJoinStop extends AsyncFunction [gps , gps_join_stop_previous]{
  14.   override def asyncInvoke(input: gps, resultFuture: ResultFuture[gps_join_stop_previous]): Unit = {
  15.     val bus_id = input.busId
  16.     val line_code = input.line_code
  17.     val stop_list = HbaseSideCache.getValueByKey("dim_hbase_line_stop" , line_code)
  18.     val bus_type = HbaseSideCache.getValueByKey("dim_hbase_bus_type" , bus_id)
  19.     val obj = Seq(gps_join_stop_previous
  20.       .apply(
  21.           input.busId,
  22.           input.runType,
  23.           input.lng,
  24.           input.lat,
  25.           input.speed,
  26.           input.mileage,
  27.           input.dataTime,
  28.           input.direction,
  29.           input.station_num,
  30.           stop_list.getString("up_end_order"),
  31.           stop_list.getString("down_end_order"),
  32.           stop_list.getString("stop_list"),
  33.           bus_type.getString("bus_type")
  34.       )
  35.     )
  36.     resultFuture.complete(obj)
  37.   }
  38. }

sink 类

  1. package com.szbus.sink
  3. import java.sql.{Connection, PreparedStatement}
  4. import com.szbus.util.MySqlClient
  5. import com.szbus.main.caseClasses.gpsSink
  6. import org.apache.flink.configuration.Configuration
  7. import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
  8. import org.apache.flink.types.Row
  9. import org.slf4j.LoggerFactory
  11. /**
  12.   *
  13.   * @author : 恋晨
  14.   *         Date  : 2019/7/3 9:06 PM
  15.   *         功能  : 自定义mysql sink类
  16.   *
  17.   */
  18. class gpsSinkToMySql  extends  RichSinkFunction[gpsSink]{
  20.   final  val log = LoggerFactory.getLogger("gpsSinkToMySql")
  21.   private var connection:Connection = _
  22.   private var ps:PreparedStatement = _
  26.   /**
  27.   *    初始化方法
  28.   *    获取数据库连接
  29.   */
  30.   override def open(parameters: Configuration): Unit ={
  31.     connection = MySqlClient.getConnection()
  32.   }
  36. /**
  37.   *    close方法
  38.   *    关闭连接释放资源
  39.   */
  40.   override def close(): Unit = {
  41.     try{
  42.       if (connection != null) {
  43.         connection.close();
  44.       }
  45.       if (ps != null) {
  46.         ps.close();
  47.       }
  49.     }catch {
  50.       case e: Exception => log.error(e.getMessage)
  52.     }
  53.   }
  56. /**
  57.   *    invoke方法
  58.   *    每条数据都会调用一次invoke方法
  59.   */  
  60.   override def invoke(value: gpsSink, context: SinkFunction.Context[_]): Unit = {
  61.     try{
  63.       val insertSql = "INSERT INTO bas_tech_bus_real_time(bus_id,run_type,lng,lat,speed,mileage,direction,next_station,event_time,car_type,comfort_level) VALUES (?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE  run_type=?,lng=?,lat=?,speed=?,mileage=?,direction=?,next_station=?,event_time=?,car_type=?,comfort_level=?;"
  65.       ps = connection.prepareStatement(insertSql)
  67.       /**组装数据，执行插入操作*/
  68.       ps.setString(1,  value.busId)
  69.       ps.setString(2,  value.runType)
  70.       ps.setString(3,  value.lng)
  71.       ps.setString(4,  value.lat)
  72.       ps.setString(5,  value.speed)
  73.       ps.setString(6,  value.mileage)
  74.       ps.setString(7,  value.direction)
  75.       ps.setString(8,  value.nextStation)
  76.       ps.setString(9,  value.dataTime)
  77.       ps.setString(10, value.bus_type)
  78.       ps.setInt(11 , value.comfort)
  79.       ps.setString(12, value.runType)
  80.       ps.setString(13, value.lng)
  81.       ps.setString(14, value.lat)
  82.       ps.setString(15, value.speed)
  83.       ps.setString(16, value.mileage)
  84.       ps.setString(17, value.direction)
  85.       ps.setString(18, value.nextStation)
  86.       ps.setString(19, value.dataTime)
  87.       ps.setString(20, value.bus_type)
  88.       ps.setInt(21 , value.comfort)
  89.       ps.executeLargeUpdate()
  91.     }catch {
  92.       case e: Exception => log.error(e.getMessage)
  93.     }
  94.   }
  95. }

flink 主类

  1. package com.szbus.main
  6. import com.alibaba.fastjson.JSON
  7. import com.szbus.join.gpsJoinStop
  8. import com.szbus.main.caseClasses.{gps,gpsSink}
  9. import com.szbus.sink.gpsSinkToMySql
  10. import com.szbus.udf.{IFF, Key2Type, getInterval, nextStation}
  11. import com.szbus.util.gpsAnalyzeFunctions._
  12. import com.szbus.source.source
  13. import org.apache.flink.api.common.time.Time
  14. import org.apache.flink.configuration.{CheckpointingOptions, Configuration}
  15. import org.apache.flink.contrib.streaming.state.{RocksDBOptions, RocksDBStateBackend}
  16. import org.apache.flink.runtime.state.StateBackend
  17. import org.apache.flink.runtime.state.filesystem.FsStateBackend
  18. import org.apache.flink.streaming.api.CheckpointingMode
  19. import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
  20. import org.apache.flink.table.api.{StreamQueryConfig, TableEnvironment}
  21. import org.apache.flink.types.Row
  22. import org.apache.flink.util.{Collector, TernaryBoolean}
  23. import org.slf4j.LoggerFactory
  25. import scala.concurrent.duration
  26. import scala.util.Random
  29. /**
  30.   *
  31.   * @author : 恋晨
  32.   *         Date  : 2019/6/28 4:59 PM
  33.   *         功能  : 解析GPS主程序
  34.   *
  35.   */
  36. object szBusGps {
  38. final  val log = LoggerFactory.getLogger("szbusGps")
  39. final  val seq = (70 to 90).map(x=>x)
  41.   def main(args: Array[String]): Unit = {
  43.     /**
  44.       * 获取flink执行环境
  45.       */
  46.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  47.     /**
  48.       * 启用checkpoint
  49.       * 设置checkpoint最小间隔为20000ms
  50.       * 设置checkpoint超时时间
  51.       * 设置任务的默认并行度为4
  52.       *
  53.       * 在数据量很大的情况下，缩短checkpoint的时间，
  54.       * 可以减小每次checkpoint 的state的大小，提高稳定性
  55.       */
  56.     env.enableCheckpointing(30000, CheckpointingMode.AT_LEAST_ONCE)
  57.     env.getCheckpointConfig.setMinPauseBetweenCheckpoints(30000)
  58.     env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
  59.     env.getCheckpointConfig.setCheckpointTimeout(900000)
  60.     env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
  61.     env.setParallelism(10)
  63.     /**
  64.       * 设置状态后端为rocksDB
  65.       *
  66.       * 线上环境建议不要使用memory 作为状态后端
  67.       * TIMER分为HEAP(默认，性能更好)和RocksDB(扩展好)
  68.       */
  70.     val checkpointDataUri ="hdfs:///flink/checkpoints-data"
  71.     val tmpDir = "file:///tmp/rocksdb/data/"
  72.     val fsStateBackend: StateBackend = new FsStateBackend(checkpointDataUri)
  73.     val rocksDBBackend: RocksDBStateBackend = new RocksDBStateBackend(fsStateBackend, TernaryBoolean.TRUE)
  74.     val config = new Configuration()
  75.     config.setString(RocksDBOptions.TIMER_SERVICE_FACTORY,RocksDBStateBackend.PriorityQueueStateType.ROCKSDB.toString)
  76.     rocksDBBackend.configure(config)
  77.     rocksDBBackend.setDbStoragePath(tmpDir)
  78.     env.setStateBackend(rocksDBBackend.asInstanceOf[StateBackend])
  80.     /**
  81.       * 获取source
  82.       */
  83.     val SOURCE = new source
  84.     val gpsSource = SOURCE.getGpsSource(env,3)
  87.     /**
  88.       *过滤掉不是车辆GPS的报文数据
  89.       * gps数据的命令字为41
  90.       */
  91.     val GpsFilterMapStream = gpsSource
  92.                .filter(x => {
  93.                             var out = false
  94.                             if (x.length >= 12 ){
  96.                               /**
  97.                                 * length从报文中截取出报文长度的编码
  98.                                 * command为报文的命令字
  99.                                 * dataLen报文长度
  100.                                 */
  102.                               val length = x.substring(4 , 8)
  103.                               val command = x.substring(10 , 12)
  104.                               val dataLen = Integer.parseInt(length , 16)
  106.                               if("41".equals(command) & x.length > 8 + dataLen*2){
  107.                                 out =   true
  108.                               }else{
  109.                                 out =   false
  110.                               }
  111.                             }else{
  112.                               out =  false
  113.                             }
  114.                             out
  115.                           }).name("过滤非gps报文")
  116.                           .setParallelism(3)
  118.                      /**
  119.                        * 报文字段拆分
  120.                        */
  121.                      .map(x =>{(
  122.                                   x.substring(42 , 58),
  123.                                   x.substring(160 , 162),
  124.                                   x.substring(110 , 120),
  125.                                   x.substring(120 , 128),
  126.                                   x.substring(144 , 148),
  127.                                   x.substring(164 , 172),
  128.                                   x.substring(156 , 158),
  129.                                   x.substring(128 , 140),
  130.                                   x.substring(58 , 74),
  131.                                   x.substring(92 , 94)
  132.                                 )}).name("拆分报文")
  133.                          .setParallelism(3)
  135.     /**
  136.       *将拆分出来的报文 转化成 具体的字段值
  137.       */
  138.     val messageSplitStream =  GpsFilterMapStream
  139.         .map(x =>{
  141.                     var speed = "0.0"
  142.                     var mileage = "0.0"
  144.                     try{
  145.                       speed = (x._5.toDouble/10).toString
  146.                       mileage = (x._6.toDouble/100).toString
  147.                     }catch{
  148.                       case e:Exception => log.error(e.getMessage)
  149.                     }
  151.                     gps(
  152.                       asciiToString(stringToDecimal(x._1)),
  153.                       if(!"1".equals(bytes2BinaryStr(hexStringToByte(x._7)).substring(6,7))) "营运中" else operationStatus2(x._2),
  154.                       DfToDd("lon" , x._3).toString,
  155.                       DfToDd("lat" , x._4).toString,
  156.                       speed,
  157.                       mileage,
  158.                       "20" + x._8,
  159.                       bytes2BinaryStr(hexStringToByte(x._7)).substring(3,4),
  160.                       asciiToString(stringToDecimal(x._9)).replaceAll("\u0000" , ""),
  161.                       Integer.parseInt(x._10 , 16).toString
  162.                     )
  163.                   }).name("根据深标协议对字段解码").setParallelism(6)
  164.       .filter(x =>{
  165.         (x.lat != "0.0" || x.lng != "0.0") && x.line_code != null && x.line_code != "" && x.station_num != null && x.station_num != "" && x.busId != null && x.busId != ""
  166.       }).name("过滤无效gps")
  167.      .setParallelism(6)
  169.     val gps_join_stop = AsyncDataStream.unorderedWait(
  170.       messageSplitStream , new gpsJoinStop , 2 , duration.MINUTES
  171.     ).name("gps数据join线路站点信息").setParallelism(10)
  173.     val gpsNextStream =  gps_join_stop
  174.       .map(x =>{
  175.         val next_Station = nextStation(x.station_num,x.up_end_order,x.down_end_order,x.stop_list)
  176.         val comfort = Random.shuffle(seq).take(1)(0)
  177.         gpsSink.apply(
  178.            x.busId,
  179.            x.runType,
  180.            x.lng,
  181.            x.lat,
  182.            x.speed,
  183.            x.mileage,
  184.            x.direction,
  185.            next_Station,
  186.            x.dataTime,
  187.            x.bus_type,
  188.           if(x.station_num== x.up_end_order || x.station_num== x.down_end_order) 100 else comfort
  189.         )
  190.     }).name("计算车辆下一站").setParallelism(10)
  192.     gpsNextStream
  193.       .addSink(new gpsSinkToMySql).name("gpsSinkToMySql").setParallelism(10)
  195.    env.execute("GPS实时任务")
  197.   }
  200.   /**
  201.     *                         计算车辆实时下一站
  202.     * @param current_order    车辆当前站点序号(上下行合在一起的序号)
  203.     * @param up_end_order     上行终点站序号
  204.     * @param down_end_order   下行终点站序号
  205.     * @param stop_list        线路站点列表,jsonString格式
  206.     * @return                 返回车辆前站点序号对应的下一站,当车辆到达上行或者下行终点站时返回 “--”
  207.     */
  208.   def nextStation (current_order:String , up_end_order:String , down_end_order:String , stop_list:String): String = {
  210.     var next_station = "--"
  211.     if(stop_list == null || current_order == null){
  212.       next_station
  213.     }else{
  214.       if(up_end_order == current_order || down_end_order == current_order){
  215.         next_station
  216.       }else{
  217.         next_station = JSON.parseObject(stop_list).getString((current_order.toInt + 1).toString)
  218.         if(next_station == null){
  219.           "--"
  220.         }else{
  221.           next_station
  222.         }
  223.       }
  224.     }
  225.   }
  226. }