Open-Source Kafka Source Stream

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

• If the Kafka server listens on the port using hostname, you need to add the mapping between the hostname and IP address of the Kafka Broker node to the DLI queue. Contact the Kafka service deployment personnel to obtain the hostname and IP address of the Kafka Broker node. For details about how to add an IP-domain mapping, see Modifying the Host Information in the Data Lake Insight User Guide.
• Kafka is an offline cluster. You need to use the enhanced datasource connection function to connect Flink jobs to Kafka. You can also set security group rules as required.

  For details about how to create 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 in the Virtual Private Cloud User Guide.

Syntax

1
2
3
4
5
6
7
8
9

CREATE SOURCE STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* )
  WITH (
    type = "kafka",
    kafka_bootstrap_servers = "",
    kafka_group_id = "",
    kafka_topic = "",
    encode = "json",
    json_config=""
  );

Keywords

Precautions

When creating a source stream, you can specify a time model for subsequent calculation. Currently, DLI supports two time models: Processing Time and Event Time. For details about the syntax, see Configuring Time Models.

Example

• Read Kafka topic test. The data encoding format is non-nested JSON, for example, {"attr1": "lilei", "attr2": 18}.

  1 2 3 4 5 6 7 8 9

  CREATE SOURCE STREAM kafka_source (name STRING, age int) WITH ( type = "kafka", kafka_bootstrap_servers = "ip1:port1,ip2:port2", kafka_group_id = "sourcegroup1", kafka_topic = "test", encode = "json", json_config = "name=attr1;age=attr2" );

• Read Kafka topic test. The data is encoded in JSON format and nested. This example uses the complex data type ROW. For details about the syntax of ROW, see Data Types.

  The test data is as follows:

  {
      "id":"1",
      "type2":"online",
      "data":{
          "patient_id":1234,
          "name":"bob1234"
      }
  }

  An example of the table creation statements is as follows:

  CREATE SOURCE STREAM kafka_source 
  (
    id STRING,
    type2 STRING,
    data ROW<
      patient_id STRING, 
      name STRING>
  )
  WITH (
    type = "kafka",
    kafka_bootstrap_servers = "ip1:port1,ip2:port2", 
    kafka_group_id = "sourcegroup1", 
    kafka_topic = "test",
    encode = "json"
  );
  
  CREATE SINK STREAM kafka_sink 
  (
    id STRING,
    type2 STRING,
    patient_id STRING, 
    name STRING
  )
    WITH (
      type="kafka",
      kafka_bootstrap_servers =  "ip1:port1,ip2:port2",
      kafka_topic = "testsink",
      encode = "csv" 
    );
  
  INSERT INTO kafka_sink select id, type2, data.patient_id, data.name from kafka_source;