IntervalJoin語句 - Realtime Compute for Apache Flink

本文為您介紹如何使用IntervalJoin語句。

背景資訊

IntervalJoin語句可以讓兩個流進行JOIN時，左流和右流中每條記錄只關聯另外一條流上滿足定義的時間範圍內的資料，且進行完JOIN後，仍然保留輸入資料流上的時間列，讓您繼續進行基於Event Time的操作。

文法格式

SELECT column-names
FROM table1  [AS <alias1>]
[INNER | LEFT | RIGHT |FULL ] JOIN table2 
ON table1.column-name1 = table2.key-name1 AND TIMEBOUND_EXPRESSION

說明

支援INNER JOIN、LEFT JOIN、RIGHT JOIN和FULL JOIN，如果直接使用JOIN，預設為INNER JOIN。
暫不支援SEMI JOIN和ANTI JOIN。
TIMEBOUND_EXPRESSION為左右兩個流時間屬性列上的區間條件運算式，支援以下三種條件運算式：
- ltime = rtime
- ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE
- ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

樣本1（基於Event Time）

統計下單後4個小時內的物流資訊。

測試資料
- 訂單表（Orders）
  id
  productName
  orderTime
  1
  iphone
  2020-04-01 10:00:00.0
  2
  mac
  2020-04-01 10:02:00.0
  3
  huawei
  2020-04-01 10:03:00.0
  4
  pad
  2020-04-01 10:05:00.0
- 物流表（Shipments）
  shipId
  orderId
  status
  shiptime
  0
  1
  shipped
  2020-04-01 11:00:00.0
  1
  2
  delivered
  2020-04-01 17:00:00.0
  2
  3
  shipped
  2020-04-01 12:00:00.0
  3
  4
  shipped
  2020-04-01 11:30:00.0

測試語句

CREATE TEMPORARY TABLE Orders(
  id BIGINT,
  productName VARCHAR,
  orderTime TIMESTAMP,
  WATERMARK wk FOR orderTime as withOffset(orderTime, 2000)  --為rowtime定義Watermark。
) WITH (
  type='datahub',
  endpoint='<yourEndpoint>',
  accessId='<yourAccessID>',
  accessKey='<yourAccessSecret>',
  projectName='<yourProjectName>',
  topic='<yourTopic>',
  project='<yourProjectName>'
);

CREATE TEMPORARY TABLE Shipments(
  shipId BIGINT,
  orderId BIGINT,
  status VARCHAR,
  shiptime TIMESTAMP,
  WATERMARK wk FOR shiptime as withOffset(shiptime, 2000)  --為rowtime定義Watermark。
) WITH (
  type='datahub',
  endpoint='<yourEndpoint>',
  accessId='<yourAccessID>',
  accessKey='<yourAccessSecret>',
  projectName='<yourProjectName>',
  topic='<yourTopic>',
  project='<yourProjectName>'
);

--使用RDS作為結果表。
CREATE TEMPORARY TABLE rds_output(
  id BIGINT,
  productName VARCHAR,
  status VARCHAR
) WITH (
  type='rds',
  url='<yourDatabaseURL>',
  tableName='<yourDatabaseTablename>',
  userName='<yourDatabaseUserName>',
  password='<yourDatabasePassword>'
);

INSERT INTO rds_output
SELECT id, productName, status
FROM Orders AS o
JOIN Shipments AS s on o.id = s.orderId AND
     o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime;

測試結果
id(bigint)
productName(varchar)
status(varchar)
1
iphone
shipped
3
huawei
shipped
4
pad
shipped

樣本2（基於Processing Time）

測試資料
- datahub_stream1
  k1
  v1
  1
  val1
  2
  val2
  3
  val3
- datahub_stream2
  k1
  v1
  1
  val1
  2
  val2
  3
  val3

測試語句

CREATE TEMPORARY TABLE datahub_stream1 (
  k1 BIGINT,
  v1 VARCHAR,
  d AS PROCTIME()
) WITH (
  'connector' = 'datahub',
  'subId' = '<yourSubId>',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'topic' = '<yourTopicName>',
  'accessId' = '<yourAccessId>',
  'accessKey' = '<yourAccessKey>'
);

CREATE TEMPORARY TABLE datahub_stream2 (
  k2 BIGINT,
  v2 VARCHAR,
  e AS PROCTIME()
) WITH (
  'connector' = 'datahub',
  'subId' = '<yourSubId>',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'topic' = '<yourTopicName>',
  'accessId' = '<yourAccessId>',
  'accessKey' = '<yourAccessKey>'
);

--使用RDS作為結果表。
CREATE TEMPORARY TABLE rds_output(
  k1 BIGINT,
  v1 VARCHAR,
  v2 VARCHAR
) WITH (
  'connector'='rds',
  'tableName'='your-table-name',
  'userName'='your-user-name',
  'password'='your-password',
  'url'='your-url'
);

INSERT INTO rds_output
SELECT k1, v1, v2
FROM datahub_stream1 AS o
JOIN datahub_stream2 AS s on o.k1 = s.k2 AND
     o.d BETWEEN s.e - INTERVAL '4' MINUTE AND s.e;

說明

由於結果取決於兩個流裡每條資料進入系統的時間，具有不確定性，因此該樣本暫不提供預期結果。

id	productName	orderTime
1	iphone	2020-04-01 10:00:00.0
2	mac	2020-04-01 10:02:00.0
3	huawei	2020-04-01 10:03:00.0
4	pad	2020-04-01 10:05:00.0

shipId	orderId	status	shiptime
0	1	shipped	2020-04-01 11:00:00.0
1	2	delivered	2020-04-01 17:00:00.0
2	3	shipped	2020-04-01 12:00:00.0
3	4	shipped	2020-04-01 11:30:00.0

id(bigint)	productName(varchar)	status(varchar)
1	iphone	shipped
3	huawei	shipped
4	pad	shipped

k1	v1
1	val1
2	val2
3	val3

k1	v1
1	val1
2	val2
3	val3