FlinkSQL Kafka Table Development Rules

Topic Must Be Specified When Kafka Is the Sink

[Example] Insert a message to the test_sink topic in Kafka.

CREATE TABLE KafkaSink(
   `user_id` VARCHAR,
   `user_name` VARCHAR,
   `age` INT 
) WITH (
   'connector' = 'kafka',
   'topic' = 'test_sink',
   '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',
   'properties.security.protocol' = 'SASL_PLAINTEXT',
   'properties.kerberos.domain.name' = 'hadoop.System domain name'
);
INSERT INTO KafkaSink (`user_id`, `user_name`, `age`)VALUES ('1', 'John Smith', 35);

properties.group.id Must Be Specified When Kafka Is the Source

[Example] Use testGroup as the user group to read Kafka messages whose topic is test_sink.

CREATE TABLE KafkaSource(
   `user_id` VARCHAR,
   `user_name` VARCHAR,
   `age` INT 
) WITH (
   'connector' = 'kafka',
   'topic' = 'test_sink',
   'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port number',
   'scan.startup.mode' = 'latest-offset',
   'properties.group.id' = 'testGroup',
   'value.format' = 'csv',
   'properties.sasl.kerberos.service.name' = 'kafka',
   'properties.security.protocol' = 'SASL_PLAINTEXT',
   'properties.kerberos.domain.name' = 'hadoop.System domain name'
);
SELECT * FROM KafkaSource;

Do Not Set Both topic-pattern and topic

topic-pattern: topic pattern, which is used for the source table. The topic name supports the regular expressions.

[Example] Subscribe to all topic messages that start with test-topic- and end with a digit for the source table:

CREATE TABLE payments ( 
    payment_id INT, 
    customer_id INT, 
    payment_date TIMESTAMP(3), 
    payment_amount DECIMAL(10, 2)
) WITH ( 
    'connector' = 'kafka', 
    'topic-pattern' = 'test-topic-[0-9]', 
    'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port number',
    'format' = 'json'
);
SELECT * FROM payments WHERE payment_amount < 500;

• 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.