Kafka

Overview

The Kafka Load Node supports to write data into Kafka topics. It can support to write data in the normal fashion and write data in the upsert fashion. The upsert-kafka connector can consume a changelog stream. It will write INSERT/UPDATE_AFTER data as normal Kafka messages value, and write DELETE data as Kafka messages with null values (indicate tombstone for the key).

Supported Version

Dependencies

In order to set up the Kafka Load Node, the following provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with Sort Connectors JAR bundles.

Maven dependency

<dependency>
    <groupId>org.apache.inlong</groupId>
    <artifactId>sort-connector-kafka</artifactId>
    <version>1.12.0</version>
</dependency>

How to create a Kafka Load Node

Usage for SQL API

The example below shows how to create a Kafka Load Node with Flink SQL :

• connector is kafka-inlong

-- Create a Kafka table 'kafka_load_node' in Flink SQL
Flink SQL> CREATE TABLE kafka_load_node (
           `id` INT,
           `name` STRINTG
           ) WITH (
           'connector' = 'kafka-inlong',
           'topic' = 'user',
           'properties.bootstrap.servers' = 'localhost:9092',
           'properties.group.id' = 'testGroup',
           'format' = 'csv'
           )

• connector is upsert-kafka

-- Create a Kafka table 'kafka_load_node' in Flink SQL
Flink SQL> CREATE TABLE kafka_load_node (
          `id` INT,
          `name` STRINTG,
           PRIMARY KEY (`id`) NOT ENFORCED
          ) WITH (
          'connector' = 'upsert-kafka-inlong',
          'topic' = 'user',
          'properties.bootstrap.servers' = 'localhost:9092',
          'key.format' = 'csv',
          'value.format' = 'csv'
          )

Usage for InLong Dashboard

When creating a data flow, select Kafka for the data stream direction, and click “Add” to configure it.

Usage for InLong Manager Client

TODO: It will be supported in the future.

Kafka Load Node Options

Available Metadata

It supports write metadata for format canal-json-inlong.

See the Kafka Extract Node for a list of all available metadata fields.

Features

Support Dynamic Schema Writing

Dynamic schema writing supports dynamic extraction of topic and partition from data and writes to the corresponding topic and partition. In order to support dynamic schema writing, you need to set the format of Kafka to ‘raw’, Also need to set the serialization format of the upstream data (via the option ‘sink.multiple.format’ to set, currently only supports [canal-json|debezium-json]).

Dynamic Topic Extraction

Dynamic topic extraction is by parsing the topic pattern and extracting the topic from the data. In order to support dynamic extraction of topic, you need to set the option ‘topic-pattern’, Kafka Load Node will parse ‘topic-pattern’ as the final topic, If parsing fails, it will be written to the default topic set via ‘topic’. ‘topic-pattern’ supports constants and variables, constants are string constants, variables are strictly represented by ‘${VARIABLE_NAME}’, and the value of the variable comes from the data itself, that is, through ‘sink.multiple.format’ a metadata field of a specified Format, or a physical field in the data.

Examples of ‘topic-parttern’ are as follows:

• ‘sink.multiple.format’ is ‘canal-json’:

The upstream data is:

{
  "data": [
    {
      "id": "111",
      "name": "scooter",
      "description": "Big 2-wheel scooter",
      "weight": "5.18"
    }
  ],
  "database": "inventory",
  "es": 1589373560000,
  "id": 9,
  "isDdl": false,
  "mysqlType": {
    "id": "INTEGER",
    "name": "VARCHAR(255)",
    "description": "VARCHAR(512)",
    "weight": "FLOAT"
  },
  "old": [
    {
      "weight": "5.15"
    }
  ],
  "pkNames": [
    "id"
  ],
  "sql": "",
  "sqlType": {
    "id": 4,
    "name": 12,
    "description": 12,
    "weight": 7
  },
  "table": "products",
  "ts": 1589373560798,
  "type": "UPDATE"
}

‘topic-pattern’ is ‘{database}_${table}’, and the extracted topic is ‘inventory_products’ (‘database’, ‘table’ are metadata fields)

‘topic-pattern’ is ‘{database}${table}${id}’, and the extracted topic is ‘inventory_products_111’ (‘database’, ‘table’ are metadata fields, and ‘id’ are physical fields)

• ‘sink.multiple.format’ is ‘debezium-json’:

The upstream data is:

{
  "before": {
    "id": 4,
    "name": "scooter",
    "description": "Big 2-wheel scooter",
    "weight": 5.18
  },
  "after": {
    "id": 4,
    "name": "scooter",
    "description": "Big 2-wheel scooter",
    "weight": 5.15
  },
  "source": {
    "db": "inventory",
    "table": "products"
  },
  "op": "u",
  "ts_ms": 1589362330904,
  "transaction": null
}

‘topic-pattern’ is ‘{source.db}_${source.table}’, and the extracted topic is ‘inventory_products’ (‘source.db’, ‘source.table’ are metadata fields)

‘topic-pattern’ is ‘{source.db}${source.table}${id}’, and the extracted topic is ‘inventory_products_4’ (‘source.db’, ‘source.table’ are metadata fields, and ‘id’ are physical fields)

Dynamic Partition Extraction

Dynamic partition extraction is to extract Partition from data by parsing partition pattern, which is similar to dynamic topic extraction. To support dynamic extraction of topics, you need to set the option ‘sink.partitioner’ to ‘raw-hash’ and option ‘sink.multiple.partition-pattern’, Kafka Load Node will parse ‘sink.multiple.partition-pattern’ as the partition key, hash the partition key and take the remainder of the partition size as the final partition, If parsing fails, it will return null and execute Kafka’s default partitioning strategy. ‘sink.multiple.partition-pattern’ support constants, variables and primary keys. Constants are string constants. Variables are strictly represented by ‘${VARIABLE_NAME}’, the value of the variable comes from the data itself, that is, it can be a metadata field of a format specified by ‘sink.multiple.format’, or it can be a physical field in the data. The primary key is a special constant ‘PRIMARY_KEY’, which extracts the primary key value of the record based on a certain format data format.

Notes: Kafka dynamic partition extraction based on ‘PRIMARY_KEY’ has a limitation that the primary key information needs to be specified in the data, For example, if Format is ‘canal-json’, then its primary key Key is ‘pkNames’. In addition, because format ‘debezium-json’ has no definition of primary key, here we agree that the primary key of ‘debezium-json’ is also ‘pkNames’ and is included in ‘source’ like other metadata fields such as ‘table’ and ‘db’, If partitioning by primary key is used, and the format is ‘debezium-json’, you need to ensure that the real data meets the above conventions.

Data Type Mapping

Kafka stores message keys and values as bytes, so Kafka doesn’t have schema or data types. The Kafka messages are deserialized and serialized by formats, e.g. csv, json, avro. Thus, the data type mapping is determined by specific formats. Please refer to Formats pages for more details.