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' ='Service IP address of the Kafka Broker instance:Kafka port',
   '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' ='Service IP address of the Kafka Broker instance:Kafka port',
   '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 single 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' = 'localhost:9092', 
    'format' = 'json'
);
SELECT * FROM payments WHERE payment_amount < 500;