Kafka Source Table

Function

Create a source stream to obtain data from Kafka as input data for jobs.

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

Kafka is an offline cluster. You have built an enhanced datasource connection to connect Flink jobs to Kafka. You have set security group rules as required.

Precautions

SASL_SSL cannot be enabled for the interconnected Kafka cluster.

Syntax

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

create table kafkaSource(
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
  (',' WATERMARK FOR rowtime_column_name AS watermark-strategy_expression)
)
with (
  'connector.type' = 'kafka',
  'connector.version' = '',
  'connector.topic' = '',
  'connector.properties.bootstrap.servers' = '',
  'connector.properties.group.id' = '',
  'connector.startup-mode' = '',
  'format.type' = ''
);

Parameters

Example

• Create table kafkaSource and read data encoded in CSV format from Kafka.

  1 2 3 4 5 6 7 8 9 10 11 12 13 14

  create table kafkaSource( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT) with ( 'connector.type' = 'kafka', 'connector.version' = '0.11', 'connector.topic' = 'test-topic', 'connector.properties.bootstrap.servers' = 'xx.xx.xx.xx:9092', 'connector.properties.group.id' = 'test-group', 'connector.startup-mode' = 'latest-offset', 'format.type' = 'csv' );

• Create table kafkaSource and read data in non-nested JSON strings from Kafka.
  Assume that the non-nested JSON strings are as follows:

  {"car_id": 312, "car_owner": "wang", "car_brand": "tang"}
  {"car_id": 313, "car_owner": "li", "car_brand": "lin"}
  {"car_id": 314, "car_owner": "zhao", "car_brand": "han"}

  You can create the table as follows:

  1 2 3 4 5 6 7 8 9 10 11 12 13 14

  create table kafkaSource( car_id STRING, car_owner STRING, car_brand STRING ) with ( 'connector.type' = 'kafka', 'connector.version' = '0.11', 'connector.topic' = 'test-topic', 'connector.properties.bootstrap.servers' = 'xx.xx.xx.xx:9092', 'connector.properties.group.id' = 'test-group', 'connector.startup-mode' = 'latest-offset', 'format.type' = 'json' );

• Create table kafkaSource and read the nested JSON data from Kafka.

  Assume that the JSON data is as follows:

  {
      "id":"1",
      "type":"online",
      "data":{
          "patient_id":1234,
          "name":"bob1234",
          "age":"Bob",
          "gmt_create":"Bob",
          "gmt_modify":"Bob"
      }
  }

  You can create the table as follows:

  CREATE table kafkaSource(
    id STRING,
    type STRING,
    data ROW(
      patient_id STRING, 
      name STRING, 
      age STRING, 
      gmt_create STRING, 
      gmt_modify STRING)
  ) 
  with (
    'connector.type' = 'kafka',
    'connector.version' = '0.11',
    'connector.topic' = 'test-topic',
    'connector.properties.bootstrap.servers' = 'xx.xx.xx.xx:9092',
    'connector.properties.group.id' = 'test-group',
    'connector.startup-mode' = 'latest-offset',
    'format.type' = 'json'
  );