Realtime Compute for Apache Flink:Jointure temporelle en temps de traitement

-- Enable processing-time temporal joins.
SET 'table.optimizer.proctime-temporal-join-strategy' = 'TEMPORAL_JOIN';
-- Disable checkpointing during full synchronization.
SET 'execution.checkpointing.interval-during-backlog' = '0';

-- Define the Kafka fact table.
-- proc_time is a computed column that returns the current processing time.
CREATE TEMPORARY TABLE kafka_input (
  id        BIGINT,
  name      VARCHAR,
  age       BIGINT,
  proc_time AS PROCTIME()
) WITH (
  'connector'                    = 'kafka',
  'topic'                        = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id'          = '<yourKafkaConsumerGroupId>',
  'format'                       = 'csv'
);

-- Define the MySQL dimension table.
CREATE TEMPORARY TABLE phoneNumber (
  name        VARCHAR,
  phoneNumber BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector'     = 'mysql',
  'hostname'      = '<yourHostname>',
  'port'          = '3306',
  'username'      = '<yourUsername>',
  'password'      = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name'    = '<yourTableName>'
);

-- Define the output table (blackhole discards results; replace with your target sink).
CREATE TEMPORARY TABLE result_infor (
  id          BIGINT,
  phoneNumber BIGINT,
  name        VARCHAR
) WITH (
  'connector' = 'blackhole'
);

-- Join the Kafka stream with the MySQL dimension table at processing time.
INSERT INTO result_infor
SELECT
  t.id,
  w.phoneNumber,
  t.name
FROM kafka_input AS t
JOIN phoneNumber FOR SYSTEM_TIME AS OF t.proc_time AS w
ON t.name = w.name;