Flink SQL Syntax Enhancement

FlinkSQL DISTRIBUTEBY

The DISTRIBUTEBY feature is added to Flink SQL to partition data based on specified fields. A single or multiple fields are supported, solving the problem where only data needs to be partitioned. The following are some examples:

SELECT /*+ DISTRIBUTEBY('id') */ id, name FROM t1;
SELECT /*+ DISTRIBUTEBY('id', 'name') */ id, name FROM t1;
SELECT /*+ DISTRIBUTEBY('id1') */ id as id1, name FROM t1;

Processing Late Data in Flink SQL Window Functions

Window functions are added to Flink SQL to support late data processing. Currently, late data is supported in the TUMBLE, HOP, OVER, and CUMULATE window functions. The following is an example:

CREATE TABLE T1 (
 `int` INT,
 `double` DOUBLE,
 `float` FLOAT,
 `bigdec` DECIMAL(10, 2),
 `string` STRING,
 `name` STRING,
 `rowtime` TIMESTAMP(3),
 WATERMARK for `rowtime` AS `rowtime` - INTERVAL '1' SECOND
) WITH ( 
 'connector' = 'values',
);

-- The fields of the sink must be the same as the input data of the window, but the sequence can be different.
CREATE TABLE LD_SINK(
 `float` FLOAT, `string` STRING, `name` STRING,  `rowtime` TIMESTAMP(3)
) WITH ( 
 'connector' = 'print',
);

SELECT  /*+ LATE_DATA_SINK('sink.name'='LD_SINK') */
  `name`,
  MIN(`float`),
  COUNT(DISTINCT `string`)
FROM TABLE(
  TUMBLE(TABLE T1, DESCRIPTOR(rowtime), INTERVAL '5' SECOND))
GROUP BY `name`, window_start, window_end

This feature also supports the output of the start time and end time of the current window when the window receives late data. The time can be output by adding window.start.field and window.end.field to the hint. The field type must be timestamp. The following is an example:

CREATE TABLE LD_SINK(
 `float` FLOAT, `string` STRING, `name` STRING,  `rowtime` TIMESTAMP(3), `windowStart` TIMESTAMP(3), `windowEnd` TIMESTAMP(3)
) WITH ( 
 'connector' = 'print',
);

SELECT  /*+ LATE_DATA_SINK('sink.name'='LD_SINK', 'window.start.field'='windowStart', 'window.end.field'='windowEnd') */
  `name`,
  MIN(`float`),
  COUNT(DISTINCT `string`)
FROM TABLE(
  TUMBLE(TABLE T1, DESCRIPTOR(rowtime), INTERVAL '5' SECOND))
GROUP BY `name`, window_start, window_end

Setting Source Parallelism

This topic is available for MRS 3.3.0 or later only.

Flink SQL allows you to use the source.parallelism parameter to set the number of concurrent source operators to deal with data skew and back pressure and improve job performance.

This feature changes the Forward partition of source and downstream operators to the Rebalance partition. When the number of concurrent source operators is different from the number of concurrent downstream operators (parallelisms) and data disorder is not allowed, enable the DISTRIBUTEBY feature together with this feature. For details, see Flink SQL Syntax Enhancement.

The following example sets the number of concurrent source operators to 2 and enables the DISTRIBUTEBY feature:

CREATE TABLE KafkaSource (
`user_id` VARCHAR,
`user_name` VARCHAR,
 `age` INT
 ) 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',
 -- Set the number of concurrent source operators.
 'source.parallelism' = '2'
 ); 
CREATE TABLE KafkaSink( 
  `user_id` VARCHAR, 
  `user_name` VARCHAR,  
 `age` INT
 ) WITH ( 
  'connector' = 'kafka', 
  'topic' = 'test_sink', 
  'properties.bootstrap.servers' = 'Service IP address of the Kafka broker instance:Kafka port', 
  'value.format' = 'csv', 
  'properties.sasl.kerberos.service.name' = 'kafka',
   'properties.security.protocol' = 'SASL_PLAINTEXT',
   'properties.kerberos.domain.name' = 'hadoop.System domain name'
 ); 
--  Insert into KafkaSink select user_id, user_name, age from KafkaSource (DISTRIBUTEBY disabled)
-- Enable DISTRIBUTEBY.
Insert into KafkaSink select/*+ DISTRIBUTEBY('user_id') */ user_id, user_name, age from KafkaSource;