Realtime Compute for Apache Flink:Jointure d'intervalle

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

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;