Flink Interval Join，Temporal Join，Lookup Join区别

Posted 2022-08-13 星辰安安

Flink Interval Join，Temporal Join，Lookup Join区别

Interval Join 间隔关联

Interval Join 多用于事件时间，如双流join中一条流关联另一条流在指定间隔时间内的记录，使用方法如下：

SELECT *
FROM Orders o, Shipments s
WHERE o.id = s.order_id
AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time

Temporal Join 时态关联

temporal join牵扯到一个很重要的概念，叫temporal table(时态表)或dynamic table(动态表) ，时态表中的行与一个或多个时态周期相关联，并且所有 Flink 表都是时态的(动态的)。时态表包含一个或多个版本化的表快照，它可以是一个跟踪变更的变更历史表(例如数据库变更日志，包含所有快照) ，也可以是一个实现变更的变更维度表(例如包含最新快照的数据库表)。

Flink SQL 流批一体的核心是：流表二象性。围绕这一核心有若干概念，例如，动态表（Dynamic Table）/时态表（Temporal Table）、版本（Version）、版本表（Version Table）、普通表、连续查询、物化视图/虚拟视图、CDC（Change Data Capture）、Changelog Stream。

• 版本： 时态表可以划分成一系列带版本的表快照集合，表快照中的版本代表了快照中所有记录的有效区间，有效区间的开始时间和结束时间可以通过用户指定，根据时态表是否可以追踪自身的历史版本与否，时态表可以分为 版本表 和 普通表。
• 版本表： 如果时态表中的记录可以追踪和并访问它的历史版本，这种表我们称之为版本表，来自数据库的 changelog 可以定义成版本表。
• 普通表： 如果时态表中的记录仅仅可以追踪并和它的最新版本，这种我们称之为普通表，来自数据库 或 HBASE的表可以定义成普通表。

Event Time Temporal Join

使用事件时间属性(即 rowtime 属性) ，可以像过去的某个时刻那样检索键的值。这允许在一个公共时间点连接两个表。版本化的表将存储自上一个水印以来的所有版本(按时间标识)。

-- Create a table of orders. This is a standard
-- append-only dynamic table.
CREATE TABLE orders (
    order_id    STRING,
    price       DECIMAL(32,2),
    currency    STRING,
    order_time  TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time
) WITH (/* ... */);

-- Define a versioned table of currency rates. 
-- This could be from a change-data-capture
-- such as Debezium, a compacted Kafka topic, or any other
-- way of defining a versioned table. 
CREATE TABLE currency_rates (
    currency STRING,
    conversion_rate DECIMAL(32, 2),
    update_time TIMESTAMP(3) METADATA FROM `values.source.timestamp` VIRTUAL,
    WATERMARK FOR update_time AS update_time,
    PRIMARY KEY(currency) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'value.format' = 'debezium-json',
   /* ... */
);

SELECT 
     order_id,
     price,
     currency,
     conversion_rate,
     order_time
FROM orders
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time
ON orders.currency = currency_rates.currency;

order_id  price  currency  conversion_rate  order_time
========  =====  ========  ===============  =========
o_001     11.11  EUR       1.14             12:00:00
o_002     12.51  EUR       1.10             12:06:00

Processing Time Temporal Join

处理时间时态表联接使用处理时间属性将行与外部版本化表中的键的最新版本相关联。处理时间时态连接最常用于使用外部表(即维度表)丰富流。

Lookup Join 查找关联

Lookup Join通常用于使用从外部系统查询的数据来丰富表。联接需要一个表具有处理时间属性，另一个表由查找源连接器支持。作用和Processing Time Temporal Join类似，写法有相同

CREATE TABLE orders (
    order_id    STRING,
    price       DECIMAL(32,2),
    total       DECIMAL(32,2),
    currency    STRING,
    order_time  TIMESTAMP(3),
    proc_time as PROC_TIME()
) WITH (/* ... */);
-- 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;