Realtime Compute for Apache Flink:Interval join

Syntax

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

Supported join types: INNER JOIN (default when JOIN is used alone), LEFT JOIN, RIGHT JOIN, and FULL JOIN. SEMI JOIN and ANTI JOIN are not supported.

TIMEBOUND_EXPRESSION must bind the timestamp of one stream to a closed interval relative to the timestamp of the other. The following expression forms are supported:

Example: match orders shipped within 4 hours

This example joins an orders stream with a shipments stream to find orders shipped within 4 hours of placement.

Test data

Orders table:

Shipments table:

SQL statements

Both source tables use a Kafka connector with a 2-second delayed watermark on the event-time column. The join condition o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime matches orders where the order time falls within 4 hours before the ship time.

CREATE TEMPORARY TABLE Orders(
  id BIGINT,
  productName VARCHAR,
  ordertime TIMESTAMP(3),
  WATERMARK wk FOR ordertime as withOffset(ordertime, 2000)  -- 2-second delayed watermark on ordertime
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE Shipments(
  shipId BIGINT,
  orderId BIGINT,
  status VARCHAR,
  shiptime TIMESTAMP(3),
  WATERMARK wk FOR shiptime as withOffset(shiptime, 2000)  -- 2-second delayed watermark on shiptime
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

-- MySQL sink
CREATE TEMPORARY TABLE rds_output(
  id BIGINT,
  productName VARCHAR,
  status VARCHAR
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

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;

Replace the following placeholders with your actual values:

Result

Order 2 (laptop) is excluded because its ship time (17:00) is more than 4 hours after the order time (10:02).