实时计算Flink版：消息队列Kafka源表

什么是Kafka源表

消息队列Kafka版是阿里云提供的分布式、高吞吐、可扩展的消息队列服务。消息队列Kafka版广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域。

Kafka源表支持连接您自建开源Apache Kafka集群，如何通过公网连接自建Apache Kafka集群详情请参见Flink全托管集群如何访问公网？。

前提条件

连接DataFlow Kafka集群

• DataFlow Kafka集群已创建。如何创建DataFlow Kafka集群，详情请参见创建DataFlow Kafka集群。
• Flink与DataFlow Kafka集群处于同一VPC内，且DataFlow Kafka集群的安全组对Flink放通。详情请参见创建和管理专有网络和安全组概述。

连接消息队列Kafka版集群

• 消息队列Kafka版集群已创建。如何创建新的消息队列Kafka版集群，详情请参见步骤三：创建资源。
• 实时计算Flink与Kafka集群处于同一VPC内，且阿里云消息队列Kafka版已对Flink开放白名单。详情请参见配置白名单。

连接自建Apache Kafka集群

• 自建Apache Kafka集群版本在0.11及以上。
• Flink与自建Apache Kafka集群之间的网络已打通。如何通过公网连接自建集群详情请参见Flink全托管集群如何访问公网？。

网络连接排查

如果您想了解更多关于Kafka客户端与服务端的连接细节，详情请参见Kafka客户端与服务端网络连接过程详解 。

使用限制

• 仅Flink计算引擎VVR 2.0.0及以上版本支持消息队列Kafka Connector。
• 消息队列Kafka Connector仅支持读取kafka 0.10及以上版本的数据。
• 消息队列Kafka Connector仅支持Kafka 2.4版本的消费者配置项，详情请参见消费者配置项。
• 仅Flink计算引擎vvr-4.0.12-flink-1.13及以上版本支持Kafka作为CTAS的同步数据源。
• 仅JSON format支持类型推导和Schema变更，其它format暂不支持。
• 在Kafka作为CTAS语句的数据源时，仅支持将JSON变更同步到Hudi和Hologres结果表。
• 仅支持Kafka中value部分的类型推导和表结构变更。如果您需要同步Kafka key部分的列，则需要您手动在DDL中进行指定。详情请参见示例三。

DDL定义

CREATE TABLE kafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING,
  `topic` STRING METADATA VIRTUAL,
  `partition` BIGINT METADATA VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'my_excellent_topic',
  'properties.bootstrap.servers' = 'mykafka:9092',
  'properties.group.id' = 'my_excellent_group'
  'format' = 'csv',
  'scan.startup.mode' = 'earliest-offset'
)

元信息列

作为CTAS数据源

WITH参数

CREATE TABLE kafkaTable (
    ...
) WITH (
    ...
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'PLAIN',
    'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";'
);

消息格式Format

每个Format都有其对应的配置项，可以直接在WITH参数中使用。Format的配置项可参见每个Format对应的 Flink社区文档。

启动模式

起始位点优先级

在以上任何一个步骤中，如果位点过期或Kafka集群发生问题等原因导致位点无效，则会使用properties.auto.offset.reset指定的策略进行位点重置。如果您未设置该配置项，则会产生异常要求您介入。

一种常见情况是使用全新的group id开始消费。首先源表会向Kafka集群查询该group的已提交位点，由于该group id是第一次使用，不会查询到有效位点，所以会通过properties.auto.offset.reset参数配置的策略进行重置。因此在使用全新group id进行消费时，必须配置properties.auto.offset.reset来指定位点重置策略。

源表位点提交

Kafka源表只在checkpoint成功后将当前消费位点提交至Kafka集群。如果您的checkpoint间隔设置较长，您在Kafka集群侧观察到的消费位点会有延迟。在进行checkpoint时，Kafka源表会将当前读取进度存储在状态中，并不依赖于提交到集群上的位点进行故障恢复，提交位点仅仅是为了在Kafka侧能够监控到读取进度，位点提交失败不会对数据正确性产生任何影响。

代码示例

• 示例一：从Kafka中读取数据后插入Kafka。
  从名称为source的Topic中读取Kafka数据，再写入名称为sink的Topic，数据使用CSV格式。

  CREATE TEMPORARY TABLE Kafka_source (
    id INT,
    name STRING,
    age INT
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'source',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'properties.group.id' = '<yourKafkaConsumerGroupId>',
    'format' = 'csv'
  );
  
  CREATE TEMPORARY TABLE Kafka_sink (
    id INT,
    name STRING,
    age INT
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'sink',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'properties.group.id' = '<yourKafkaConsumerGroupId>',
    'format' = 'csv'
  );
  
  INSERT INTO Kafka_sink SELECT id, name, age FROM Kafka_source;

• 示例二：同步表结构以及数据
  将Kafka Topic中的消息实时同步到Hologres中。在该情况下，您可以将Kafka消息的offset和partition id作为主键，从而保证在Failover时，Hologres中不会有重复消息。

  CREATE TEMPORARY TABLE kafkaTable (
    `offset` INT NOT NULL METADATA,
    `part` BIGINT NOT NULL METADATA FROM 'partition',
    PRIMARY KEY (`part`, `offset`) NOT ENFORCED
  ) WITH (
    'connector' = 'kafka',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'topic' = 'kafka_evolution_demo',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json',
    'json.infer-schema.flatten-nested-columns.enable' = 'true'
      -- 可选，将嵌套列全部展开。
  );
  
  CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
  WITH (
    'connector' = 'hologres'
  ) AS TABLE vvp.`default`.kafkaTable;

• 示例三：同步表结构以及Kafka消息的key和value数据。
  Kafka消息中的key部分已经存储了相关信息，您可以同时同步Kafka中的key和value。

  CREATE TEMPORARY TABLE kafkaTable (
    `key_id` INT NOT NULL,
    `val_name` VARCHAR(200)
  ) WITH (
    'connector' = 'kafka',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'topic' = 'kafka_evolution_demo',
    'scan.startup.mode' = 'earliest-offset',
    'key.format' = 'json',
    'value.format' = 'json',
    'key.fields' = 'key_id',
    'key.fields-prefix' = 'key_',
    'value.fields-prefix' = 'val_',
    'value.fields-include' = 'EXCEPT_KEY'
  );
  
  CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
  WITH (
    'connector' = 'hologres'
  ) AS TABLE vvp.`default`.kafkaTable;

  说明 Kafka消息中的key部分不支持表结构变更和类型推导，需要您手动声明。
• 示例四：同步表结构和数据并进行计算。
  在同步Kafka数据到Hologres时，往往需要一些轻量级的计算。

  CREATE TEMPORARY TABLE kafkaTable (
    `distinct_id` INT NOT NULL,
    `properties` STRING,
    `timestamp` TIMESTAMP METADATA,
    `date` AS CAST(`timestamp` AS DATE)
  ) WITH (
    'connector' = 'kafka',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'topic' = 'kafka_evolution_demo',
    'scan.startup.mode' = 'earliest-offset',
    'key.format' = 'json',
    'value.format' = 'json',
    'key.fields' = 'key_id',
    'key.fields-prefix' = 'key_'
  );
  
  CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
     'connector' = 'hologres'
  ) AS TABLE vvp.`default`.kafkaTable
  ADD COLUMN
    `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
  -- 使用COALESCE处理空值情况。

常见问题

• Flink如何获取JSON数据？
• Flink和Kafka网络连通，但Flink无法消费或者写入数据？
• Flink中的Commit Offset有什么作用？
• 为什么Flink和Kafka之间的网络是连通的，但是依然会有timeout expired while fetching topic metadata的报错？
• 为什么Kafka源表数据基于Event Time的窗口后，不能输出数据？