Help Center/ MapReduce Service/ Component Operation Guide (LTS)/ Using Flink/ Enterprise-Class Enhancements of Flink/ Using PROCTIME() Syntax in Flink SQL Lookup Join
Updated on 2026-06-27 GMT+08:00
View PDF
Share

Using PROCTIME() Syntax in Flink SQL Lookup Join

This section applies only to MRS 3.6.0-LTS or later.

Scenarios

In the Flink Lookup Join syntax, FOR SYSTEM_TIME AS OF must include the proc_time field, for example, FOR SYSTEM_TIME AS OF o.proc_time. Now, you can use PROCTIME() to replace the proc_time field in the table and directly use FOR SYSTEM_TIME AS OF PROCTIME(). The proc_time field no longer needs to be defined in the table. This is also supported by subqueries.

How to Use

When you configure a Flink job, Lookup Join allows direct use of PROCTIME() to replace the proc_time field in the table.

Example SQL statements:

CREATE TABLE kafkasource(
  order_id STRING,
  price DECIMAL(32, 2),
  currency STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'test_source',
  'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'csv',
  'properties.sasl.kerberos.service.name' = 'kafka',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.kerberos.domain.name' = 'hadoop.System domain name'
);
CREATE TEMPORARY TABLE customers (id STRING, name STRING, country STRING) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://mysqlhost:ip/customerdb',
  'table-name' = 'customers',
  'username' = 'xxxx',
  'password' = 'xxxx'
);
CREATE TABLE print (
  order_id STRING,
  price DECIMAL(32, 2),
  name STRING,
  country STRING
) WITH ('connector' = 'print');
insert into
  print
SELECT
  o.order_id,
  o.price,
  c.name,
  c.country
FROM
  kafkasource as o
  JOIN customers FOR SYSTEM_TIME AS OF PROCTIME() AS c ON o.order_id = c.id;