Kafka

Overview

The Kafka Extract Node supports to read data from Kafka topics. It can support read data in the normal fashion and read data in the upsert fashion. The upsert-kafka connector produces a changelog stream, where each data record represents an update or delete event. The kafka-inlong connector can read data and metadata.

Supported Version

Dependencies

In order to set up the Kafka Extract 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.8.0</version>
</dependency>

How to create a Kafka Extract Node

Usage for SQL API

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

• connector is kafka-inlong

-- Set checkpoint every 3000 milliseconds                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';   
-- Create a Kafka table 'kafka_extract_node' in Flink SQL
Flink SQL> CREATE TABLE kafka_extract_node (
           `id` INT,
           `name` STRINTG
           ) WITH (
           'connector' = 'kafka-inlong',
           'topic' = 'user',
           'properties.bootstrap.servers' = 'localhost:9092',
           'properties.group.id' = 'testGroup',
           'scan.startup.mode' = 'earliest-offset',
           'format' = 'csv'
           )
-- Read data
Flink SQL> SELECT * FROM kafka_extract_node;

• connector is upsert-kafka

-- Set checkpoint every 3000 milliseconds                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';
-- Create a Kafka table 'kafka_extract_node' in Flink SQL
Flink SQL> CREATE TABLE kafka_extract_node (
          `id` INT,
          `name` STRINTG,
           PRIMARY KEY (`id`) NOT ENFORCED
          ) WITH (
          'connector' = 'upsert-kafka-inlong',
          'topic' = 'user',
          'properties.bootstrap.servers' = 'localhost:9092',
          'properties.group.id' = 'testGroup',
          'scan.startup.mode' = 'earliest-offset',
          'key.format' = 'csv',
          'value.format' = 'csv'
          )
-- Read data
Flink SQL> SELECT * FROM kafka_extract_node;

Usage for InLong Dashboard

TODO: It will be supported in the future.

Usage for InLong Manager Client

TODO: It will be supported in the future.

Kafka Extract Node Options

Available Metadata

The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition. It supports read metadata for format canal-json-inlong.

The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:

CREATE TABLE `kafka_extract_node` (
      `id` INT,
      `name` STRING,
      `database_name` string METADATA FROM 'value.database_name',
      `table_name`    string METADATA FROM 'value.table_name',
      `op_ts`         timestamp(3) METADATA FROM 'value.op_ts',
      `op_type` string METADATA FROM 'value.op_type',
      `batch_id` bigint METADATA FROM 'value.batch_id',
      `is_ddl` boolean METADATA FROM 'value.is_ddl',
      `update_before` ARRAY<MAP<STRING, STRING>> METADATA FROM 'value.update_before',
      `mysql_type` MAP<STRING, STRING> METADATA FROM 'value.mysql_type',
      `pk_names` ARRAY<STRING> METADATA FROM 'value.pk_names',
      `data` STRING METADATA FROM 'value.data',
      `sql_type` MAP<STRING, INT> METADATA FROM 'value.sql_type',
      `ingestion_ts` TIMESTAMP(3) METADATA FROM 'value.ts',
) WITH (
      'connector' = 'kafka-inlong',
      'topic' = 'user',
      'properties.bootstrap.servers' = 'localhost:9092',
      'properties.group.id' = 'testGroup',
      'scan.startup.mode' = 'earliest-offset',
      'format' = 'canal-json-inlong'
)

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.