Kafka/BMQ

Kafka 连接器提供从 Kafka Topic 或 BMQ Topic 中消费和写入数据的能力，支持做数据源表和结果表。您可以创建 source 流从 Kafka Topic 中获取数据，作为作业的输入数据；也可以通过 Kafka 结果表将作业输出数据写入到 Kafka Topic 中。

注意事项

• 使用 Flink SQL 的用户需要注意，不再支持 kafka-0.10 和 kafka-0.11 两个版本的连接器，请直接使用 kafka 连接器访问 Kafka 0.10 和 0.11 集群。
  Kafka-0.10 和 Kafka-0.11 两个版本的连接器使用的 Kafka 客户端有缺陷，在某些情况下可能无法自动提交 Kafka offset 信息。
• 使用 datastream API 开发的用户需要注意，在读 Kafka 消息的时候，不要使用 FlinkKafkaConsumer010 和 FlinkKafkaConsumer011 两个 consumer，请直接使用 FlinkKafkaConsumer 进行开发；在往 Kafka 写消息的时候，不要使用 FlinkKafkaProducer010 和 FlinkKafkaProducer011 两个 producer，请直接使用 FlinkKafkaProducer 进行开发。

DDL 定义

用作数据源（Source）

CREATE TABLE kafka_source (
    name String,
    score INT
 ) WITH (
     'connector' = 'kafka',
     'topic' = 'test_topic_01',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'test_topic_01',
     'format' = 'csv',
     'scan.startup.mode' = 'earliest-offset'
 );

用作数据目的（Sink）

CREATE TABLE kafka_sink (
    name String,
    score INT
 ) WITH (
     'connector' = 'kafka',
     'topic' = 'test_topic_01',
     'properties.bootstrap.servers' = 'localhost:9092',
     'format' = 'csv'
 );

WITH 参数

自动提交 Offsets

目前一般使用以下两种方式自动提交 Kafka Offsets。

• 方式 1：依赖 Flink 任务 Checkpoint。
  Flink 任务开启 Checkpoint 时，Kafka Source 在完成 Checkpoint 时会提交当前的消费位点，以保证 Flink 的 Checkpoint 状态和 Kafka Broker 上的提交位点一致。

  注意

  依赖 Flink 任务 Checkpoint 来管理 Kafka Offsets 时，如果上游数据量很大，很可能会触发上游的 LAG 告警。

• 方式 2：依赖 Kafka Consumer 的位点定时提交逻辑。
  当 Flink 任务没有开启 Checkpoint 时，Kafka Source 将依赖 Kafka Consumer 的位点定时提交逻辑。您可以通过设置enable.auto.commit和auto.commit.interval.ms两个参数来控制位点定时自动提交。

  -- 是否自动提交 Offsets。取值为 true 表示自动提交 Offsets；取值为 false，表示手动同步或异步提交。
    'enable.auto.commit' = 'true',
    -- 自动提交 Offsets 的时间间隔，单位为 ms。
    'auto.commit.interval.ms' = '500',
  

安全与认证

如果 Kafka 集群要求安全连接或认证，您需要在 WITH 参数中通过 properties.前缀添加安全认证相关配置。
示例 1：使用 SASL_PLAINTEXT 安全协议，SASL 机制为 PLAIN 。

CREATE TABLE KafkaTable (
  user_id BIGINT,
  item_id BIGINT,
  behavior STRING
) WITH (
  'connector' = 'kafka',
  ...
  -- 配置安全协议为 SASL_PLAINTEXT。
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  -- 配置 SASL 机制为  PLAIN。
  'properties.sasl.mechanism' = 'PLAIN',
  -- 配置 JAAS。
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="yourname" password="yourpassword";'
)

示例 2：使用 SASL_SSL 安全协议， SASL 机制为 SASL_SSL，并配置证书和密钥信息。

CREATE TABLE KafkaTable (
  user_id BIGINT,
  item_id BIGINT,
  behavior STRING
) WITH (
  'connector' = 'kafka',
  ...
  -- 配置安全协议为 SASL_SSL。
  'properties.security.protocol' = 'SASL_SSL',
  -- 配置服务端提供的 truststore (CA 证书) 的路径和密码。
  'properties.ssl.truststore.location' = '/path/to/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  -- 如果要求客户端认证，则需要配置 keystore (私钥) 的路径和密码。
  'properties.ssl.keystore.location' = '/path/to/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  --配置 SASL 机制为 SCRAM-SHA-256。 
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  -- 配置 JAAS。
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="yourname" password="yourpassword";'
)

示例代码

• 源表示例

  CREATE TABLE kafka_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp
       )
  WITH (
    'connector' = 'kafka',
    'topic' = 'source',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'group_01',  -- 消费BMQ消息时，需要提前在BMQ平台侧创建Group，否则不能正常提交Offset。
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
  );
  CREATE TABLE print_sink (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp
      )
  WITH (
       'connector' = 'print'           
  );
  insert into print_sink
  select * from kafka_source;
  

• 结果表示例

  CREATE TABLE datagen_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time as localtimestamp
      )
  WITH (
       'connector' = 'datagen',
       'rows-per-second' = '5'          
  );
  
  CREATE TABLE kafka_sink (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time timestamp
       )
  WITH (
    'connector' = 'kafka',
    'topic' = 'sink',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'group_01',
    'format' = 'json'
  );
  
  insert into kafka_sink
  select * from datagen_source;