5.5、Flink Table API & SQL - 连接外部系统之 Kafka Connector（Kafka连接器）

朴灿烈づ我的快乐病毒、 2024-04-17 06:28 65阅读 0赞

Kafka连接器允许从Apache Kafka主题读写内容。它可以定义为：

**JAVA / SCALA**

.connect(
      new Kafka()
        .version("0.11")    // required: valid connector versions are
                            //   "0.8", "0.9", "0.10", "0.11", and "universal"
        .topic("...")       // required: topic name from which the table is read
    
        // optional: connector specific properties
        .property("zookeeper.connect", "localhost:2181")
        .property("bootstrap.servers", "localhost:9092")
        .property("group.id", "testGroup")
    
        // optional: select a startup mode for Kafka offsets
        .startFromEarliest()
        .startFromLatest()
        .startFromSpecificOffsets(...)
    
        // optional: output partitioning from Flink's partitions into Kafka's partitions
        .sinkPartitionerFixed()         // each Flink partition ends up in at-most one Kafka partition (default)
        .sinkPartitionerRoundRobin()    // a Flink partition is distributed to Kafka partitions round-robin
        .sinkPartitionerCustom(MyCustom.class)    // use a custom FlinkKafkaPartitioner subclass
    )

**DDL**

CREATE TABLE MyUserTable (
      ...
    ) WITH (
      'connector.type' = 'kafka',       
    
      'connector.version' = '0.11',     -- required: valid connector versions are
                                        -- "0.8", "0.9", "0.10", "0.11", and "universal"
    
      'connector.topic' = 'topic_name', -- required: topic name from which the table is read
    
      'update-mode' = 'append',         -- required: update mode when used as table sink, 
                                        -- only support append mode now.
    
      'connector.properties.0.key' = 'zookeeper.connect', -- optional: connector specific properties
      'connector.properties.0.value' = 'localhost:2181',
      'connector.properties.1.key' = 'bootstrap.servers',
      'connector.properties.1.value' = 'localhost:9092',
      'connector.properties.2.key' = 'group.id',
      'connector.properties.2.value' = 'testGroup',
      'connector.startup-mode' = 'earliest-offset',    -- optional: valid modes are "earliest-offset", 
                                                       -- "latest-offset", "group-offsets", 
                                                       -- or "specific-offsets"
    
      -- optional: used in case of startup mode with specific offsets
      'connector.specific-offsets.0.partition' = '0',
      'connector.specific-offsets.0.offset' = '42',
      'connector.specific-offsets.1.partition' = '1',
      'connector.specific-offsets.1.offset' = '300',
    
      'connector.sink-partitioner' = '...',  -- optional: output partitioning from Flink's partitions 
                                             -- into Kafka's partitions valid are "fixed" 
                                             -- (each Flink partition ends up in at most one Kafka partition),
                                             -- "round-robin" (a Flink partition is distributed to 
                                             -- Kafka partitions round-robin)
                                             -- "custom" (use a custom FlinkKafkaPartitioner subclass)
      -- optional: used in case of sink partitioner custom
      'connector.sink-partitioner-class' = 'org.mycompany.MyPartitioner'
    )

**指定开始读取位置：**默认情况下，Kafka source 将开始从 Zookeeper 或 Kafka broker 中提交的组的偏移量读取数据。您可以指定其他启动位置，这些位置对应于 Kafka 消费者启动位置配置节中的配置，详情请查阅：[Flink流处理（Stream API）- Connector（连接器） 之 Apache Kafka Connector][Flink_Stream API_- Connector_ _ Apache Kafka Connector]。

**Flink-Kafka Sink 分区：**默认情况下，Kafka接收器最多只写它自己的并行度(接收器的每个并行实例只写一个分区)。可以自定义 sink 分区来将数据写到多个分区中。轮询分区可以将数据均匀的发给每个分区。但是，它将导致所有 Flink 实例和所有 Kafka 代理之间的大量网络连接。

**一致性：**默认情况下，如果查询是在启用 checkpoint 的情况下执行的，Kafka 接收器会将带有至少一次语义保证的数据摄取到 Kafka 主题中。

**Kafka0.10+时间戳：**从 Kafka 0.10开始, Kafka 消息有一个时间戳作为元数据，指定记录何时写入 Kafka 主题。通过分别选择时间戳，YAML 中的 from-source 和 Java/Scala 中的 timestampsFromSource()，可以将这些时间戳用于 rowtime 属性。

**Kafka0.11+版本：**从Flink 1.7开始,，Kafka 连接器定义应该独立于 hard-coded 的 Kafka 版本。连接器版本 universal 作为 Flink 的 Kafka 连接器的通配符，兼容从0.11开始的所有Kafka版本。确保添加特定于版本的Kafka依赖项。此外，还需要指定相应的格式来读写 Kafka 的数据。

[Flink_Stream API_- Connector_ _ Apache Kafka Connector]: https://blog.csdn.net/zpf_940810653842/article/details/100511937#%E6%B6%88%E8%B4%B9%E8%80%85%E5%BC%80%E5%A7%8B%E6%B6%88%E8%B4%B9%E4%BD%8D%E7%BD%AE%E9%85%8D%E7%BD%AE