手册   游戏   WEB   字典   单词   在线工具  

Flink 常见维表 Join 方案, 收藏学习开发很有用!

由于维表是一张不断变化的表(静态表只是动态表的一种特例). 那如何 JOIN 一张不断变化的表呢? 如果用传统的 JOIN 语法来表达维表 JOIN, 是不完整的. 因为维表是一直在更新变化的, 如果用这个语法那么关联上的是哪个时刻的维表呢? 我们是不知道的, 结果是不确定的.

前言

实时数仓, 难免会遇到 join 维表的业务. 现总结几种方案, 供各位看官选择:

查找关联(同步, 异步)

状态编程, 预加载数据到状态中, 按需取

冷热数据

广播维表

  1. Temporal Table Join
  2. Lookup Table Join

其中中间留下两个问题, 供大家思考, 可留言一起讨论?

查找关联

查找关联就是在主流数据中直接访问外部数据 (MySQL,Redis,impala ...) 去根据主键或者某种关键条件去关联取值.

适合: 维表数据量大, 但是主数据不大的业务实时计算.

缺点: 数据量大的时候, 会给外部数据源库带来很大的压力, 因为某条数据都需要关联.

同步

访问数据库是同步调用, 导致 subtak 线程会被阻塞, 影响吞吐量

  1. import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
  2. import com.wang.stream.env.{FlinkStreamEnv, KafkaSourceEnv}
  3. import org.apache.flink.API.common.functions.FlatMapFunction
  4. import org.apache.flink.API.common.serialization.SimpleStringSchema
  5. import org.apache.flink.streaming.API.scala._
  6. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
  7. import org.apache.flink.util.Collector
  8.  def analyses(): Unit ={
  9.     val env: StreamExecutionEnvironment = FlinkStreamEnv.get()
  10.     KafkaSourceEnv.getKafkaSourceStream(env,List("test"))
  11.       .map(JSON.parseObject(_))
  12.       .filter(_!=null)
  13.       .flatMap(
  14.         new FlatMapFunction[JSONObject,String] {
  15.           override def flatMap(jSONObject: JSONObject, collector: Collector[String]): Unit = {
  16.             // 如果 topic 就一张表, 不用区分, 如果多张表, 可以通过 database 与 table 区分, 放到下一步去处理
  17.             // 表的名字
  18.             val databaseName:String = jSONObject.getString("database")
  19.             // 表的名字
  20.             val tableName:String = jSONObject.getString("table")
  21.             // 数据操作类型 INSERT UPDATE DELETE
  22.             val operationType:String = jSONObject.getString("type")
  23.             // 主体数据
  24.             val tableData: JSONArray = jSONObject.getJSONArray("data")
  25.             // old 值
  26.             val old: JSONArray = jSONObject.getJSONArray("old")
  27.             // canal JSON 可能存在批处理出现 data 数据多条
  28.             for (i <- 0 until tableData.size()) {
  29.               val data: String = tableData.get(i).toString
  30.               val nObject: JSONObject = JSON.parseObject(data)
  31.               val orderId: AnyRef = nObject.get("order_id")
  32.               // 下面写 (MySQL,Redis 或者 hbase) 的连接, 利用 API 通过 orderId 查找
  33.               // 最后封装数据格式 就是 join 所得
  34.               collector.collect(null)
  35.             }
  36.           }
  37.         }
  38.       )
  39.       .addSink(
  40.         new FlinkKafkaProducer[String](
  41.           "",
  42.           "",
  43.           new SimpleStringSchema()
  44.         )
  45.       )
  46.     env.execute("")

异步

AsyncIO 可以并发地处理多个请求, 很大程度上减少了对 subtask 线程的阻塞.

  1. def analyses(): Unit ={
  2.     val env: StreamExecutionEnvironment = FlinkStreamEnv.get()
  3.     val source: DataStream[String] = KafkaSourceEnv.getKafkaSourceStream(env, List("test"))
  4.       .map(JSON.parseObject(_))
  5.       .filter(_ != null)
  6.       .flatMap(
  7.         new FlatMapFunction[JSONObject, String] {
  8.           override def flatMap(jSONObject: JSONObject, collector: Collector[String]): Unit = {
  9.             // 如果 topic 就一张表, 不用区分, 如果多张表, 可以通过 database 与 table 区分, 放到下一步去处理
  10.             // 表的名字
  11.             val databaseName: String = jSONObject.getString("database")
  12.             // 表的名字
  13.             val tableName: String = jSONObject.getString("table")
  14.             // 数据操作类型 INSERT UPDATE DELETE
  15.             val operationType: String = jSONObject.getString("type")
  16.             // 主体数据
  17.             val tableData: JSONArray = jSONObject.getJSONArray("data")
  18.             // old 值
  19.             val old: JSONArray = jSONObject.getJSONArray("old")
  20.             // canal JSON 可能存在批处理出现 data 数据多条
  21.             for (i <- 0 until tableData.size()) {
  22.               val data: String = tableData.get(i).toString
  23.               collector.collect(data)
  24.             }
  25.           }
  26.         }
  27.       )
  28.     AsyncDataStream.unorderedWait(
  29.       source,
  30.       new RichAsyncFunction[String,String] {// 自定义的数据源异步处理类
  31.         override def open(parameters: Configuration): Unit = {
  32.           // 初始化
  33.         }
  34.         override def asyncInvoke(input: String, resultFuture: ResultFuture[String]): Unit = {
  35.           // 将数据搜集
  36.           resultFuture.complete(null)
  37.        }
  38.         override def close(): Unit = {
  39.           // 关闭
  40.         }
  41.     },
  42.     1000,// 异步超时时间
  43.     TimeUnit.MILLISECONDS,// 时间单位
  44.     100)// 最大异步并发请求数量
  45.      .addSink(
  46.         new FlinkKafkaProducer[String](
  47.           "",
  48.           "",
  49.           new SimpleStringSchema()
  50.         )
  51.       )
  52.     env.execute("")
  53.   }

状态编程, 预加载数据到状态中, 按需取

首先把维表数据初始化到 state 中, 设置好更新时间, 定时去把维表.

优点: flink 自己维护状态数据,"荣辱与共", 不需要频繁链接外部数据源, 达到解耦.

缺点: 不适合大的维表和变化大的维表.

  1. .keyBy(_._1)
  2. .process(
  3.   new KeyedProcessFunction[String,(String,String,String,String,String), String]{
  4.     private var mapState:MapState[String,Map[String,String]] = _
  5.     private var first: Boolean = true
  6.     override def open(parameters: Configuration): Unit = {
  7.       val config: StateTtlConfig = StateTtlConfig
  8.         .newBuilder(org.apache.flink.API.common.time.Time.minutes(5))
  9.         .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
  10.         .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
  11.         .build()
  12.       val join = new MapStateDescriptor[String,Map[String,String]]("join",classOf[String],classOf[Map[String,String]])
  13.       join.enableTimeToLive(config)
  14.       mapState = getRuntimeContext.getMapState(join)
  15.     }
  16.     override def processElement(
  17.                                  in: (String, String, String, String, String),
  18.                                  context: KeyedProcessFunction[String, (String, String, String, String, String),String]#Context,
  19.                                  collector: Collector[String]): Unit = {
  20.       // 加载维表
  21.       if(first){
  22.         first = false
  23.         val time: Long = System.currentTimeMillis()
  24.         getSmallDimTableInfo()
  25.         // 设置好更新时间, 定时去把维表
  26.         context.timerService().registerProcessingTimeTimer(time + 86400000)
  27.       }
  28.       // 数据处理, 过来一条条数据, 然后按照自己的业务逻辑去取维表的数据即可
  29.       // 然后封装 放到 collect 中
  30.       collector.collect(null)
  31.     }
  32.     override def onTimer(
  33.                           timestamp: Long,
  34.                           ctx: KeyedProcessFunction[String, (String, String, String, String, String),String]#OnTimerContext,
  35.                           out: Collector[String]): Unit = {
  36.       println("触发器执行")
  37.       mapState.clear()
  38.       getSmallDimTableInfo()
  39.       println(mapState)
  40.       ctx.timerService().registerProcessingTimeTimer(timestamp + 86400000)
  41.     }
  42.     def getSmallDimTableInfo(): Unit ={
  43.       // 加载 字典数据
  44.       val select_dictionary="select dic_code,pre_dictionary_id,dic_name from xxxx"
  45.       val dictionary: util.List[util.Map[String, AnyRef]] = MysqlUtil.executeQuery(select_dictionary, null)
  46.       dictionary.foreach(item=>{
  47.         mapState.put("dic_dictionary_"+item.get("pre_dictionary_id").toString,item)
  48.       })
  49.     }
  50.   }
  51. )
  52. .filter(_!=null)
  53. .addSink(
  54.   new FlinkKafkaProducer[String](
  55.     "",
  56.     "",
  57.     new SimpleStringSchema()
  58.   )
  59. )
  60. v.execute("")

思考下: 直接定义一个 Map 集合这样的优缺点是什么? 可以留言说出自己的看法?

冷热数据

思想: 先去状态去取, 如果没有, 去外部查询, 同时去存到状态里面. StateTtlConfig 的过期时间可以设置短点.

优点: 中庸取值方案, 热备常用数据到内存, 也避免了数据 join 相对过多外部数据源.

缺点: 也不能一劳永逸解决某些问题, 热备数据过多, 或者冷数据过大, 都会对 state 或者 外部数据库造成压力.

  1. .filter(_._1 != null)
  2. .keyBy(_._1)
  3. .process(
  4.   new KeyedProcessFunction[String,(String,String,String,String,String), String]{
  5.     private var mapState:MapState[String,Map[String,String]] = _
  6.     private var first: Boolean = true
  7.     override def open(parameters: Configuration): Unit = {
  8.       val config: StateTtlConfig = StateTtlConfig
  9.         .newBuilder(org.apache.flink.API.common.time.Time.days(1))
  10.         .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
  11.         .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
  12.         .build()
  13.       val join = new MapStateDescriptor[String,Map[String,String]]("join",classOf[String],classOf[Map[String,String]])
  14.       join.enableTimeToLive(config)
  15.       mapState = getRuntimeContext.getMapState(join)
  16.     }
  17.     override def processElement(
  18.                                  in: (String, String, String, String, String),
  19.                                  context: KeyedProcessFunction[String, (String, String, String, String, String),String]#Context,
  20.                                  collector: Collector[String]): Unit = {
  21.       // 数据处理, 过来一条条数据, 然后按照自己的业务逻辑先去 mapState 去找, 如果没有再去 外部去找
  22.       if (mapState.contains("xx_id")){
  23.         // 如果存在就取
  24.       }else{
  25.         // 如果不存在去外部拿, 然后放到 mapState 中
  26.         val dim_sql="select dic_code,pre_dictionary_id,dic_name from xxxx where id=xx_id"
  27.         val dim: util.List[util.Map[String, AnyRef]] = MysqlUtil.executeQuery(dim_sql, null)
  28.         mapState.put("xx_id",null)
  29.       }
  30.       // 然后封装 放到 collect 中
  31.       collector.collect(null)
  32.     }
  33.   }
  34. )

广播维表

比如上面提到的字典表, 每一个 Task 都需要这份数据, 那么需要 join 这份数据的时候就可以使用广播维表.

  1. val dimStream=env.addSource(MysqlSource)
  2. // 广播状态
  3. val broadcastStateDesc=new MapStateDescriptor[String,String]("broadcaststate", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(Long.class, Dim.class))
  4. // 广播流
  5. val broadStream=dimStream.broadcast()
  6. // 主数据流
  7. val mainConsumer = new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), kafkaConfig)
  8. val mainStream=env.addSource(mainConsumer)
  9. // 广播状态与维度表关联
  10. val connectedStream=mainStream.connect(broadStream).map(..User(id,name)).key(_.1)
  11. connectedStream.process(new KeyedBroadcastProcessFunction[String,User,Map[Long,Dim],String] {
  12.      override def processElement(value: User, ctx: KeyedBroadcastProcessFunction[String,User,Map[Long,Dim],String]#ReadOnlyContext, out: Collector[String]): Unit = {
  13.       // 取到数据就可以愉快的玩耍了
  14.      val state=ctx.getBroadcastState(broadcastStateDesc)
  15.        xxxxxx
  16.   }
  17. }

「思考:」 如果把维表流也通过实时监控 binlog 到 kafka, 当维度数据发生变化时, 更新放到状态中, 这种方式, 是不是更具有时效性呢?

(1)通过 canal 把变更 binlog 方式发送到 kafka 中.

(2)数据流定义成为广播流, 广播到数据到主数据流中.

(3)定义一个广播状态存储数据, 在主数据进行查找匹配, 符合要求则 join 成功.

Temporal Table Join(FlinkSQL 与 Flink Table API)

由于维表是一张不断变化的表(静态表只是动态表的一种特例). 那如何 JOIN 一张不断变化的表呢? 如果用传统的 JOIN 语法来表达维表 JOIN, 是不完整的. 因为维表是一直在更新变化的, 如果用这个语法那么关联上的是哪个时刻的维表呢? 我们是不知道的, 结果是不确定的. 所以 Flink SQL 的维表 JOIN 语法引入了 Temporal Table 的标准语法, 用来声明关联的是维表哪个时刻的快照.

普通关联会一直保留关联双侧的数据, 数据也就会一直膨胀, 直到撑爆内存导致任务失败, Temporal Join 则可以定期清理过期数据, 在合理的内存配置下即可避免内存溢出.

Event Time Temporal Join

语法

  1. SELECT [column_list]
  2. FROM table1 [AS ]
  3. [LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{
  4. 	proctime | rowtime	
  5. } [AS ]
  6. ON table1.column-name1 = table2.column-name1

使用事件时间属性(即行时间属性), 可以检索过去某个时间点的键值. 这允许在一个共同的时间点连接两个表.

举例

假设我们有一个订单表, 每个订单都有不同货币的价格. 为了将此表正确地规范化为单一货币, 每个订单都需要与下订单时的适当货币兑换率相结合.

  1. CREATE TABLE orders (
  2.     order_id    STRING,
  3.     price       DECIMAL(32,2),
  4.     currency    STRING,
  5.     order_time  TIMESTAMP(3),
  6.     WATERMARK FOR order_time AS order_time
  7. ) WITH (/* ... */);
  8. CREATE TABLE currency_rates (
  9.     currency STRING,
  10.     conversion_rate DECIMAL(32, 2),
  11.     update_time TIMESTAMP(3) METADATA FROM `values.source.timestamp` VIRTUAL
  12.     WATERMARK FOR update_time AS update_time,
  13.     PRIMARY KEY(currency) NOT ENFORCED
  14. ) WITH (
  15.    'connector' = 'upsert-kafka',
  16.    /* ... */
  17. );

event-time temporal join 需要 temporal join 条件的等价条件中包含的主键

  1. SELECT
  2.      order_id,
  3.      price,
  4.      currency,
  5.      conversion_rate,
  6.      order_time,
  7. FROM orders
  8. LEFT JOIN currency_rates FOR SYSTEM TIME AS OF orders.order_time
  9. ON orders.currency = currency_rates.currency
  10. Processing Time Temporal Join

处理时间时态表连接使用处理时间属性将行与外部版本表中键的最新版本相关联.

根据定义, 使用 processing-time 属性, 连接将始终返回给定键的最新值. 可以将查找表看作是一个简单的 HashMap, 它存储来自构建端的所有记录. 这种连接的强大之处在于, 当在 Flink 中无法将表具体化为动态表时, 它允许 Flink 直接针对外部系统工作.

使用 FOR SYSTEM_TIME AS OF table1.proctime 表示当左边表的记录与右边的维表 join 时, 只匹配当前处理时间维表所对应的的快照数据.

Lookup Table Join

Lookup Join 通常用于通过连接外部表 (维度表) 补充信息, 要求一个表具有处理时间属性, 另一个表使 Lookup Source Connector.

JDBC 连接器可以用在时态表关联中作为一个可 lookup 的 source (又称为维表). 用到的语法是 Temporal Joins 的语法.

  1. s"""
  2.         |CREATE TABLE users(
  3.         |id int,
  4.         |name string,
  5.         |PRIMARY KEY (id) NOT ENFORCED
  6.         |)
  7.         |WITH (
  8.         |'connector' = 'jdbc',
  9.         |'url' = 'xxxx',
  10.         |'driver'='$DRIVER_CLASS_NAME',
  11.         |'table-name'='$tableName',
  12.         |'lookup.cache.max-rows'='100',
  13.         |'lookup.cache.ttl'='30s'
  14.         |)
  15.         |""".stripMargin
  16. s"""
  17.        |CREATE TABLE car(
  18.        |`id`   bigint ,
  19.        |`user_id` bigint,
  20.        |`proctime` as PROCTIME()
  21.        |)
  22.        |WITH (
  23.        |    'connector' = 'kafka',
  24.        |    'topic' = '$topic',
  25.        |    'scan.startup.mode' = 'latest-offset',
  26.        |    'properties.bootstrap.servers' = '$KAFKA_SERVICE',
  27.        |    'properties.group.id' = 'indicator',
  28.        |    'format' = 'canal-json'
  29.        |)
  30.        |""".stripMargin
  31.     SELECT
  32.         mc.user_id user_id,
  33.         count(1) AS `value`
  34.     FROM car mc
  35.         inner join users FOR SYSTEM_TIME AS OF mc.proctime as u on mc.user_id=s.id
  36.     group by mc.user_id

总结

总体来讲, 关联维表有四个基础的方式:

(1)查找外部数据源关联

(2)预加载维表关联(内存, 状态)

(3)冷热数据储备(算是 1 和 2 的结合使用)

(4)维表变更日志关联(广播也好, 其他方式的流关联也好)

「同时考虑:」 吞吐量, 时效性, 外部数据源的负载, 内存资源, 解耦性等等方面.

四种 join 方式不存在绝对的一劳永逸, 更多的是针对业务场景在各指标上的权衡取舍, 因此看官需要结合场景来选择适合的.

来源: http://bigdata.51cto.com/art/202201/698686.htm


与本文相关文章

  1. 深度学习中常见优化算法学习笔记
  2. Python 自动化开发学习 14
  3. 总结《精通数据科学》机器学习思想, 常见误区, 二元分类问题
  4. 零基础学习前端开发, 如何有效的学习以及学习中常见的误区!
  5. GitHub 最著名的 20 个 Python 机器学习项目, 值得收藏!
  6. os.path.join()
  7. 机器学习数学基础: 常见分布与假设检验
  8. 07-numpy - 笔记 - join
暂无,快来抢沙发吧！
更多





黑ICP备17005318号-1