Kafka

概述

Kafka Extract 节点 支持从 Kafka topics 中读取数据。它支持以普通的方式读取数据和 Upsert 的方式读取数据。upsert-kafka 连接器生产 changelog 流, 其中每条数据记录代表一个更新或删除事件。kafka-inlong 连接器可以以普通方式读取数据和元数据信息。

支持的版本

依赖

为了设置 Kafka Extract 节点, 下面提供了使用构建自动化工具（例如 Maven 或 SBT）和带有 Sort Connector JAR 包的 SQL 客户端的两个项目的依赖关系信息。

Maven 依赖

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

如何创建 Kafka Extract 节点

SQL API 用法

下面这个例子展示了如何用 Flink SQL 创建一个 Kafka Extract 节点:

• 连接器是 kafka-inlong

-- 设置 Checkpoint 为 3000 毫秒                      
Flink SQL> SET 'execution.checkpointing.interval' = '3s';   
-- 使用 Flink SQL 创建 Kafka 表 'kafka_extract_node'
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'
           )
-- 读取数据
Flink SQL> SELECT * FROM kafka_extract_node;

• 连接器是 upsert-kafka

-- 设置 Checkpoint 为 3000 毫秒                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';
-- 使用 Flink SQL 创建 Kafka 表 'kafka_extract_node'
Flink SQL> CREATE TABLE kafka_extract_node (
          `id` INT,
          `name` STRINTG,
           PRIMARY KEY (`id`) NOT ENFORCED
          ) WITH (
          'connector' = 'upsert-kafka',
          'topic' = 'user',
          'properties.bootstrap.servers' = 'localhost:9092',
          'properties.group.id' = 'testGroup',
          'scan.startup.mode' = 'earliest-offset',
          'key.format' = 'csv',
          'value.format' = 'csv'
          )
-- 读取数据
Flink SQL> SELECT * FROM kafka_extract_node;

InLong Dashboard 用法

TODO: 将在未来支持此功能。

InLong Manager Client 用法

TODO: 将在未来支持此功能。

Kafka Extract 节点参数

可用的元数据字段

以下格式元数据可以作为表定义中的只读 (VIRTUAL) 列公开。 它支持读取格式 canal-json-inlong 的元数据。

扩展的 CREATE TABLE 示例演示了使用这些元数据字段的语法：

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'
)

数据类型映射

Kafka 将消息键值以二进制进行存储，因此 Kafka 并不存在 schema 或数据类型。Kafka 消息使用格式配置进行序列化和反序列化，例如 csv，json，avro。 因此，数据类型映射取决于使用的格式。请参阅 格式 页面以获取更多细节。