FlinkSQL Kafka Table Development Suggestions

Traffic Limiting Must Be Set When Kafka Is the Source

This rule is available for MRS 3.3.0 or later.

To prevent job exceptions caused by heavy traffic, set a traffic limit, which should be the peak value of the pressure test for service rollout.

[Example] Set the traffic limit when Kafka is the source table.

# The following parameter takes effect at any parallelism:
'scan.records-per-second.limit' = '1000'
# The actual traffic limit is as follows:
min( parallelism * scan.records-per-second.limit, partitions num * scan.records-per-second.limit)

Write Data with the Same Key to the Same Kafka Partition for Data Accuracy

Flink uses the fixed policy to write data to Kafka and performs hash calculation based on the key before writing data.

[Example] Flink writes data to Kafka using the fixed policy.

CREATE TABLE kafka (
 f_sequence INT,
 f_sequence1 INT,
 f_sequence2 INT,
 f_sequence3 INT 
 ) WITH ( 
 'connector' = 'kafka',
 'topic' = 'yxtest123',
 'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port number',
 'properties.group.id' = 'testGroup1',
 'scan.startup.mode' = 'latest-offset',
 'format' = 'json',
 'sink.partitioner'='fixed'
 ); 

 insert into kafka select /*+ DISTRIBUTEBY('f_sequence','f_sequence1') */ * from datagen;

Set Kafka Source Parallelism Same as the Number of Topic Partitions for Faster Kafka Consumption

When the parallelism of Kafka Source is greater than the number of topic partitions, no more data is consumed.

Connecting to Kafka with Kerberos Authentication Enabled in the MRS Cluster (in Security Mode)

The WITH attribute of a Flink job must contain the following configurations:

#Kafka Kerberos service name
'properties.sasl.kerberos.service.name' = 'kafka',
#Protocol used for communication between the Kafka servers and clients 
'properties.security.protocol' = 'SASL_PLAINTEXT',
#System domain name: You can log in to FusionInsight Manager, choose System > Permission > Domain and Mutual Trust, and check the value of Local Domain.
'properties.kerberos.domain.name' = 'hadoop.System domain name'

[Example] Connect to Kafka topic test_kafka with Kerberos authentication enabled.

CREATE TABLE kafka (
 f_sequence INT,
 f_sequence1 INT,
 f_sequence2 INT,
 f_sequence3 INT 
 ) WITH ( 
'connector' = 'kafka',
'topic' = 'test_kafka',
'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port number',
'scan.startup.mode' = 'latest-offset',
'value.format' = 'csv',
'properties.sasl.kerberos.service.name' = 'kafka', --Kafka Kerberos service name
'properties.security.protocol' = 'SASL_PLAINTEXT', --Communication protocol used by Kafka to release ports externally
'properties.kerberos.domain.name' = 'hadoop.System domain name'
 ); 

• To obtain the IP address of the Kafka broker instance, log in to Manager, choose Cluster > Services > Kafka > Instances.
• To obtain the Kafka port number,
  • If Kerberos authentication is enabled for the cluster (the cluster is in security mode), the broker port number is the value of sasl.port, which is 21007 by default.
  • If Kerberos authentication is disabled for the cluster (the cluster is in normal mode), the broker port number is the value of port, which is 9092 by default. If the port number is set to 9092, set allow.everyone.if.no.acl.found to true. The procedure is as follows:

    Log in to FusionInsight Manager and choose Cluster > Services > Kafka. Click Configurations and then All Configurations. On the displayed page, search for allow.everyone.if.no.acl.found, set it to true, and click Save.

• To obtain the System domain name, log in to FusionInsight Manager, choose System > Permission > Domain and Mutual Trust, and record the value of Local Domain.