flink的regular join/window join/interval join/temporal join/lookup join

特点

• 左右流都会触发结果更新，而不是通过窗口结束来触发(从下面的datastream api/flink sql可以看到，并没有用诸如tumblewindow等，所以不是窗口触发计算的，而是来一条就触发计算一次）
• State 自动清理，根据时间区间保留数据
• 只支持双流，不支持三流及以上。
• Flink同时支持基于EventTime和ProcessingTime的Interval join,但是只支持append-only流。
• Interval join 已经支持inner ,left outer, right outer , full outer 等类型的join

  例如下面的left join：

  insert into realtime_db.test_window_tab
  select  t1.region
          ,t1.qa_id
  from    realtime_db.test__dwm_qa_score t1
  left join realtime_db.score_result_tab t2
  on      t1.qa_id = t2.newrow.qa_id
  where   TO_TIMESTAMP(FROM_UNIXTIME(t1.first_submit_timestamp))  between 
  TO_TIMESTAMP(FROM_UNIXTIME(t2.newrow.first_submit_time)) - interval  '1' HOUR
    and   TO_TIMESTAMP(FROM_UNIXTIME(t2.newrow.first_submit_time))
  

  

  Temporal Join 

  为什么需要temporal join:在 regular join和interval join中，join 两侧的表是平等的，任意的一个表的更新，都会去和另外的历史纪录进行匹配。而有时候，我们需要一个表 的更新对另一表在该时间节点以前的记录是不可见的，即不影响历史数据

  时态表（Temporal table）是一个随时间变化的表：在 Flink 中被称为动态表。时态表中的行与一个或多个时间段相关联，所有 Flink 中的表都是时态的（Temporal）。 时态表包含一个或多个版本的表快照，它可以是一个变化的历史表，跟踪变化（例如，数据库变化日志，包含所有快照）或一个变化的维度表，也可以是一个将变更物化的维表（例如，存放最终快照的数据表）。

  这里可以分为 JOIN 当前表和 JOIN 历史表。

  JOIN 当前维表:处理时间temporal join

  SELECT *
  FROM Orders AS o
  [LEFT] JOIN Products FOR SYSTEM_TIME AS OF PROCTIME() AS p
  ON o.productId = p.productId
  
  

  Flink SQL 支持 LEFT JOIN 和 INNER JOIN 的维表关联。如上语法所示的，维表 JOIN 语法与传统的 JOIN 语法并无二异，加了left就是left  join,否则就是inner join。

  JOIN 历史维表:事件时间temporal join

  SELECT *

  FROM Orders AS o

  [LEFT] JOIN Products FOR SYSTEM_TIME AS OF o.orderTime AS p

  ON o.productId = p.productId

  有时候想关联上的维度数据，并不是当前时刻的值，而是某个历史时刻的值(可以理解为事件时间temporal join)。比如，产品的价格一直在发生变化，订单流希望补全的是下单时的价格，而不是当前的价格，那就是 JOIN 历史维表。语法上只需要将上文的 PROCTIME()  改成  o.orderTime 即可。含义是关联上的是下单时刻的 Products 维表。

  Lookup Join

  为什么需要lookup join:因为temporal join需要双流，而有时候维表只是一个非流（例如mysql,hbase)

  lookup join 通常用于使用从外部系统查询的数据来丰富表。join 要求一个表具有处理时间属性，另一个表由查找源连接器（lookup source connnector）支持。

  lookup join 和上面的 处理时间 Temporal Join 语法相同，右表使用查找源连接器支持。

  下面的例子展示了 lookup join 的语法。

  -- Customers is backed by the JDBC connector and can be used for lookup joins
  CREATE TEMPORARY TABLE Customers (
    id INT,
    name STRING,
    country STRING,
    zip STRING
  ) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',
    'table-name' = 'customers'
  );
   
  -- enrich each order with customer information
  SELECT o.order_id, o.total, c.country, c.zip
  FROM Orders AS o
    JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
      ON o.customer_id = c.id;
  

  在上面的示例中，Orders 表由保存在 MySQL 数据库中的 Customers 表数据来丰富。带有后续 process time 属性的 FOR SYSTEM_TIME AS OF 子句确保在联接运算符处理 Orders 行时，Orders 的每一行都与 join 条件匹配的 Customer 行连接。它还防止连接的 Customer 表在未来发生更新时变更连接结果。lookup join 还需要一个强制的相等连接条件（不能空），在上面的示例中是 o.customer_id = c.id。

  维表join优化的方法

  非同步io+批量拉取

  Async 极大地提高了吞吐，但是每一次 IO 请求只取了单 key 的数据，效率比较低。未来计划使用 Batch Get 来提高每次 IO 请求的吞吐。

  缓存

  数据库的维表查询请求，有大量相同 key 的重复请求。如何减少重复请求？本地缓存是常用的方案。Flink SQL 目前提供两种缓存方案：LRU 和 ALL。

  引入 Partitioned-ALL-cache

  也就是上游数据到 JoinTable 节点根据 JOIN key 分区，那么每个节点只需要加载属于该分区key的缓存数据，从而做到了缓存的水平扩展。从而遇到超大维表时可以通过扩并发也能够全量缓存下维表数据。

  Side Input

  ALL cache 现在每个节点是都会起一个线程去加载全量维表数据，如果有1000个节点，则会全量读数据库1000次。未来打算通过 Side Input功能做到只需要全量读取一次，维表数据会自动分发到各个节点。