Kafka Result Table

Function

DLI outputs the Flink job output data to Kafka through the Kafka result table.

Apache Kafka is a fast, scalable, and fault-tolerant distributed message publishing and subscription system. It delivers high throughput and built-in partitions and provides data replicas and fault tolerance. Apache Kafka is applicable to scenarios of handling massive messages.

Prerequisites

• You have created a Kafka cluster.
• An enhanced datasource connection has been created for DLI to connect to Kafka clusters, so that jobs can run on the dedicated queue of DLI and you can set the security group rules as required.
  • For details about how to set up an enhanced datasource connection, see Enhanced Datasource Connections in the Data Lake Insight User Guide.
  • For details about how to configure security group rules, see Security Group Overview in the Virtual Private Cloud User Guide.
• In Flink cross-source development scenarios, there is a risk of password leakage if datasource authentication information is directly configured. You are advised to use the datasource authentication provided by DLI.

  For details about datasource authentication, see Introduction to Datasource Authentication.

Precautions

• When creating a Flink OpenSource SQL job, you need to set Flink Version to 1.12 on the Running Parameters tab of the job editing page, select Save Job Log, and set the OBS bucket for saving job logs.
• For details about how to use data types, see section Format.

Syntax

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

create table kafkaSink(
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
)
with (
  'connector' = 'kafka',
  'topic' = '',
  'properties.bootstrap.servers' = '',
  'format' = ''
);

Parameters

Example (SASL_SSL Disabled for the Kafka Cluster)

In this example, data is read from a Kafka topic and written to another using a Kafka result table.

1. Create an enhanced datasource connection in the VPC and subnet where Kafka locates, and bind the connection to the required Flink elastic resource pool. For details, see Enhanced Datasource Connections.
2. Set Kafka security groups and add inbound rules to allow access from the Flink queue. Test the connectivity using the Kafka address by referring to Testing Address Connectivity. If the connection is successful, the datasource is bound to the queue. Otherwise, the binding fails.
3. Create a Flink OpenSource SQL job. Enter the following job script and submit the job.
   When you create a job, set Flink Version to 1.12 on the Running Parameters tab. Select Save Job Log, and specify the OBS bucket for saving job logs. Change the values of the parameters in bold as needed in the following script.

   CREATE TABLE kafkaSource (
     order_id string,
     order_channel string,
     order_time string, 
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'KafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     'scan.startup.mode' = 'latest-offset',
     "format" = "json"
   );
   
   CREATE TABLE kafkaSink (
     order_id string,
     order_channel string,
     order_time string, 
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'KafkaSinkTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     "format" = "json"
   );
   
   insert into kafkaSink select * from kafkaSource;

4. Connect to the Kafka cluster and insert the following test data into the source topic in Kafka:

   {"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"pay_time":"2021-03-24 10:02:03","user_id":"0001","user_name":"Alice","area_id":"330106"}
   
   {"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}

5. Connect to the Kafka cluster and read data from the sink topic of Kafka.

   {"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"pay_time":"2021-03-24 10:02:03","user_id":"0001","user_name":"Alice","area_id":"330106"}
   
   {"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}

Example (SASL_SSL Enabled for the Kafka Cluster)

• Example 1: Enable SASL_SSL authentication for the DMS cluster.

  Create a Kafka cluster for DMS, enable SASL_SSL, download the SSL certificate, and upload the downloaded certificate client.jks to an OBS bucket.

  CREATE TABLE ordersSource (
    order_id string,
    order_channel string,
    order_time timestamp(3),
    pay_amount double,
    real_pay double,
    pay_time string,
    user_id string,
    user_name string,
    area_id string
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'xx',
    'properties.bootstrap.servers' = 'xx:9093,xx:9093,xx:9093',
    'properties.group.id' = 'GroupId',
    'scan.startup.mode' = 'latest-offset',
    'properties.connector.auth.open' = 'true',
    'properties.ssl.truststore.location' = 'obs://xx/xx.jks',  -- Location where the user uploads the certificate to
    'properties.sasl.mechanism' = 'PLAIN',  --  Value format: SASL_PLAINTEXT
    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=\"xx\" password=\"xx\";', -- Account and password set when the Kafka cluster is created
    "format" = "json"
  );
   
  CREATE TABLE ordersSink (
    order_id string,
    order_channel string,
    order_time timestamp(3),
    pay_amount double,
    real_pay double,
    pay_time string,
    user_id string,
    user_name string,
    area_id string
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'xx',
    'properties.bootstrap.servers' = 'xx:9093,xx:9093,xx:9093',
    'properties.connector.auth.open' = 'true',
    'properties.ssl.truststore.location' = 'obs://xx/xx.jks',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=\"xx\" password=\"xx\";',
    "format" = "json"
  );
   
  insert into ordersSink select * from ordersSource;

• Example 2: Enable Kafka SASL_SSL authentication for the MRS cluster.
  • Enable Kerberos authentication for the MRS cluster.
  • Click the Components tab and click Kafka. In the displayed page, click the Service Configuration tab, locate the security.protocol, and set it to SASL_SSL.
  • Download the user credential. Log in to the FusionInsight Manager of the MRS cluster and choose System > Permission > User. Locate the row that contains the target user, click More, and select Download Authentication Credential.

    Obtain the truststore.jks file using the authentication credential and store the credential and truststore.jks file in OBS.

  • If "Message stream modified (41)" is displayed, the JDK version may be incorrect. Change the JDK version in the sample code to a version earlier than 8u_242 or delete the renew_lifetime = 0m configuration item from the krb5.conf configuration file.
  • Set the port to the sasl_ssl.port configured in the Kafka service configuration.
  • In the following statements, set security.protocol to SASL_SSL.

  CREATE TABLE ordersSource (
    order_id string,
    order_channel string,
    order_time timestamp(3),
    pay_amount double,
    real_pay double,
    pay_time string,
    user_id string,
    user_name string,
    area_id string
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'xx',
    'properties.bootstrap.servers' = 'xx:21009,xx:21009',
    'properties.group.id' = 'GroupId',
    'scan.startup.mode' = 'latest-offset',
    'properties.sasl.kerberos.service.name' = 'kafka',
    'properties.connector.auth.open' = 'true',
    'properties.connector.kerberos.principal' = 'xx', --Username
    'properties.connector.kerberos.krb5' = 'obs://xx/krb5.conf',
    'properties.connector.kerberos.keytab' = 'obs://xx/user.keytab',
    'properties.security.protocol' = 'SASL_SSL',
    'properties.ssl.truststore.location' = 'obs://xx/truststore.jks',
    'properties.ssl.truststore.password' = 'xx',  -- Password set for generating truststore.jks
    'properties.sasl.mechanism' = 'GSSAPI',
    "format" = "json"
  );
   
  CREATE TABLE ordersSink (
    order_id string,
    order_channel string,
    order_time timestamp(3),
    pay_amount double,
    real_pay double,
    pay_time string,
    user_id string,
    user_name string,
    area_id string
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'xx',
    'properties.bootstrap.servers' = 'xx:21009,xx:21009',
    'properties.sasl.kerberos.service.name' = 'kafka',
    'properties.connector.auth.open' = 'true',
    'properties.connector.kerberos.principal' = 'xx',
    'properties.connector.kerberos.krb5' = 'obs://xx/krb5.conf',
    'properties.connector.kerberos.keytab' = 'obs://xx/user.keytab',
    'properties.ssl.truststore.location' = 'obs://xx/truststore.jks',
    'properties.ssl.truststore.password' = 'xx',
    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.mechanism' = 'GSSAPI',
    "format" = "json"
  );
   
  insert into ordersSink select * from ordersSource;

• Example 3: Enable Kerberos SASL_PAINTEXT authentication for the MRS cluster
  • Enable Kerberos authentication for the MRS cluster.
  • Click the Components tab and click Kafka. In the displayed page, click the Service Configuration tab, locate the security.protocol, and set it to SASL_PLAINTEXT.
  • Log in to the FusionInsight Manager of the MRS cluster and download the user credential. Choose System > Permission > User. Locate the row that contains the target user, choose More > Download Authentication Credential. Upload the credential to OBS.
  • If error message "Message stream modified (41)" is displayed, the JDK version may be incorrect. Change the JDK version in the sample code to a version earlier than 8u_242 or delete the renew_lifetime = 0m configuration item from the krb5.conf configuration file.
  • Set the port to the sasl.port configured in the Kafka service configuration.
  • In the following statements, set security.protocol to SASL_PLAINTEXT.

  CREATE TABLE ordersSources (
    order_id string,
    order_channel string,
    order_time timestamp(3),
    pay_amount double,
    real_pay double,
    pay_time string,
    user_id string,
    user_name string,
    area_id string
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'xx',
    'properties.bootstrap.servers' = 'xx:21007,xx:21007',
    'properties.group.id' = 'GroupId',
    'scan.startup.mode' = 'latest-offset',
    'properties.sasl.kerberos.service.name' = 'kafka',
    'properties.connector.auth.open' = 'true',
    'properties.connector.kerberos.principal' = 'xx',
    'properties.connector.kerberos.krb5' = 'obs://xx/krb5.conf',
    'properties.connector.kerberos.keytab' = 'obs://xx/user.keytab',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'GSSAPI',
    "format" = "json"
  );
   
  CREATE TABLE ordersSink (
    order_id string,
    order_channel string,
    order_time timestamp(3),
    pay_amount double,
    real_pay double,
    pay_time string,
    user_id string,
    user_name string,
    area_id string
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'xx',
    'properties.bootstrap.servers' = 'xx:21007,xx:21007',
    'properties.sasl.kerberos.service.name' = 'kafka',
    'properties.connector.auth.open' = 'true',
    'properties.connector.kerberos.principal' = 'xx',
    'properties.connector.kerberos.krb5' = 'obs://xx/krb5.conf',
    'properties.connector.kerberos.keytab' = 'obs://xx/user.keytab',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'GSSAPI',
    "format" = "json"
  );
   
  insert into ordersSink select * from ordersSource;

• Example 4: Use SSL for the MRS cluster
  • Do not enable Kerberos authentication for the MRS cluster.
  • Download the user credential. Log in to the FusionInsight Manager of the MRS cluster and choose System > Permission > User. Locate the row that contains the target user, click More, and select Download Authentication Credential.

    Obtain the truststore.jks file using the authentication credential and store the credential and truststore.jks file in OBS.

  • Set the port to the ssl.port configured in the Kafka service configuration.
  • In the following statements, set security.protocol to SSL.
  • Set ssl.mode.enable to true.

    CREATE TABLE ordersSource (
      order_id string,
      order_channel string,
      order_time timestamp(3),
      pay_amount double,
      real_pay double,
      pay_time string,
      user_id string,
      user_name string,
      area_id string
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'xx',
      'properties.bootstrap.servers' = 'xx:9093,xx:9093,xx:9093',
      'properties.group.id' = 'GroupId',
      'scan.startup.mode' = 'latest-offset',
      'properties.connector.auth.open' = 'true',
      'properties.ssl.truststore.location' = 'obs://xx/truststore.jks',
      'properties.ssl.truststore.password' = 'xx',  -- Password set for generating truststore.jks
      'properties.security.protocol' = 'SSL',
      "format" = "json"
    );
     
    CREATE TABLE ordersSink (
      order_id string,
      order_channel string,
      order_time timestamp(3),
      pay_amount double,
      real_pay double,
      pay_time string,
      user_id string,
      user_name string,
      area_id string
    ) WITH (
      'connector' = 'print'
    );
     
    insert into ordersSink select * from ordersSource;