彻底明白Flink系统学习27：【Flink1.7】连续查询连接（时态表等）

问题导读

1.Flink有哪些连接？
2.常规连接、窗口连接、时态表连接有什么区别？
3.时态表函数如何使用？

上一篇彻底明白Flink系统学习26：【Flink1.7】Table API 和SQL API之时间属性
http://www.aboutyun.com/forum.php?mod=viewthread&tid=26646


连接是批处理数据处理中常见且易于理解的操作，用于连接两个关系的行。 但是，动态表上的连接语义不太明显甚至令人困惑。

因此，有几种方法可以使用Table API或SQL实际执行连接。

常规连接
常规joins是最通用的join类型，其中任何新记录或对join输入任一侧的更改都是可见的并且正在影响整个联接（join）结果。 例如，如果左侧有新记录，则它将与右侧的所有先前和未来记录一起加入。
[mw_shl_code=sql,true]SELECT * FROM Orders
INNER JOIN Product
ON Orders.productId = Product.id[/mw_shl_code]
这些语义允许任何类型的更新（插入，更新，删除）输入表。
但是，此操作具有重要意义：它需要将连接输入的两侧永久保持在Flink的状态。 因此，如果一个或两个输入表不断增长，资源使用也将无限增长。

时间窗口连接
时间窗口连接由join 定义，该join 检查输入记录的时间属性是否在某些时间限制内，即时间窗口。
[mw_shl_code=sql,true]SELECT *
FROM
  Orders o,
  Shipments s
WHERE o.id = s.orderId AND
      o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime[/mw_shl_code]
与常规连接操作相比，此类连接仅支持具有时间属性的仅追加表（append-only tables）。 由于时间属性是quasi-monontic增加，Flink可以从其状态中删除旧值而不影响结果的正确性。

时态表连接
具有时态表的连接将仅追加（ append-only）表（左输入/探测侧）与时态表（右输入/构建侧）连接，即，随时间变化并跟踪其变化的表。 有关时态表，后面会进一步介绍。
以下示例显示了一个仅追加（append-only）表的订单，这些订单应与不断变化的货币汇率表RatesHistory连接。

Orders是仅追加（ append-only）表，表示给定金额和给定货币的付款。 例如，在10:15，订单金额为2欧元。
[mw_shl_code=sql,true]SELECT * FROM Orders;

rowtime amount currency
======= ====== =========
10:15        2 Euro
10:30        1 US Dollar
10:32       50 Yen
10:52        3 Euro
11:04        5 US Dollar[/mw_shl_code]
RatesHistory代表一个不断变化的仅追加货币汇率表，相对于日元（汇率为1）。 例如，欧元兑日元从09:00到10:45的汇率为114.从10:45到11:15，汇率为116。
[mw_shl_code=sql,true]SELECT * FROM RatesHistory;

rowtime currency   rate
======= ======== ======
09:00   US Dollar   102
09:00   Euro        114
09:00   Yen           1
10:45   Euro        116
11:15   Euro        119
11:49   Pounds      108[/mw_shl_code]
鉴于我们想要计算所有订单的金额转换为通用货币（日元）。

例如，我们希望使用转换率给定的rowtime转换如下订单 （114）

[mw_shl_code=sql,true]rowtime amount currency
======= ====== =========
10:15        2 Euro[/mw_shl_code]
如果不使用时态表，就需要编写如下查询：
[mw_shl_code=sql,true]SELECT
  SUM(o.amount * r.rate) AS amount
FROM Orders AS o,
  RatesHistory AS r
WHERE r.currency = o.currency
AND r.rowtime = (
  SELECT MAX(rowtime)
  FROM RatesHistory AS r2
  WHERE r2.currency = o.currency
  AND r2.rowtime <= o.rowtime);[/mw_shl_code]借助时态表函数Rates，可以使用下面sql查询
[mw_shl_code=sql,true]SELECT
  o.amount * r.rate AS amount
FROM
  Orders AS o,
  LATERAL TABLE (Rates(o.rowtime)) AS r
WHERE r.currency = o.currency[/mw_shl_code]来自探测器（追加）侧的每个记录将在探测器（追加）侧记录的相关时间属性时与构建侧表（的版本）连接。 为了支持构建端表上先前值的更新（覆盖），该表必须定义主键。

在我们的示例中，Orders表中的每条记录将与Rates表join，在时间o.rowtime 。 货币（currency）字段已被定义为之前的Rates的主键，并用于连接我们示例中的两个表。 如果查询使用处理时间概念，则在执行操作时，新添加的订单将始终与最新版本的Rates连接。

与常规连接相比，这意味着如果构建端（时态表）有新记录，则不会影响以前的连接结果。 这再次允许Flink限制元素的数量，保持state。

与时间窗口连接相比，时态表连接不定义时间窗口（时间窗口内的数据将会被join）。 探针端的记录始终与time属性指定的构建端版本连接。 因此，构建方面的记录可能是任意旧的。 随着时间的推移，将从状态中删除先前和不再需要的记录版本（对于给定的主键）。

用法
定义时态表函数后，我们可以使用它。 时态表函数的使用方式与使用普通表函数的方式相同。

以下代码段解决了我们从Orders表转换货币问题：
[mw_shl_code=sql,true]SELECT
  SUM(o_amount * r_rate) AS amount
FROM
  Orders,
  LATERAL TABLE (Rates(o_proctime))
WHERE
  r_currency = o_currency[/mw_shl_code]

[mw_shl_code=java,true]Table result = orders
    .join(new Table(tEnv, "rates(o_proctime)"), "o_currency = r_currency")
    .select("(o_amount * r_rate).sum as amount");[/mw_shl_code]

[mw_shl_code=scala,true]val result = orders
    .join(rates('o_proctime), 'r_currency === 'o_currency)
    .select(('o_amount * 'r_rate).sum as 'amount)[/mw_shl_code]
注意：尚未为时态连接实现查询配置中定义的状态保留。 这意味着计算查询结果所需的状态可能会无限增长，具体取决于历史记录表的不同主键的数量。

处理时间时态连接
使用处理时间属性，不可能将过去的时间属性作为参数传递给时态表函数。 根据定义，它始终是当前时间戳。 因此，处理时间时间表函数的调用将始终返回基础表的最新已知版本，并且基础历史表中的任何更新也将立即覆盖当前值。

只有构建端记录的最新版本（相对于已定义的主键）保留在该状态中。 构建端的更新将不会影响先前发出的连接结果。

可以将处理时间时间连接视为一个简单的HashMap <K，V>，它存储来自构建方的所有记录。 当构建端的新记录具有与先前记录相同的key时，旧值只是被覆盖。 始终根据HashMap的最新/当前状态评估探测器端的每条记录。

事件时间时态连接
利用事件时间属性（即rowtime属性），可以将过去的时间属性传递给时间表函数。这允许在共同的时间点连接两个表。

与处理时间时间连接相比，时态表不仅保持状态中的构建侧记录的最新版本（相对于定义的主键），而且存储自上一个水印以来的所有版本（由时间标识）。

例如，根据时态表的概念，在时间12:30:00将附加到探测器侧表的事件时间时间戳为12:30:00的传入行与构建侧表的版本连接在一起。 。因此，传入行仅与时间戳小于或等于12:30:00的行连接，并根据主键应用更新，直到此时为止。

通过事件时间的定义，水印允许连接操作及时向前移动并丢弃不再需要的构建表的版本，因为不期望具有较低或相等时间戳的传入行。

最新经典文章，欢迎关注公众号

感谢分享