连接器概览

下表列出了 Managed Service for Apache Kafka 支持的 Kafka Connect 连接器类型。您可以使用这些连接器将 Apache Kafka 与您的应用以及其他 Google Cloud服务集成。

转化者数量

转换器负责 Kafka 记录数据的序列化和反序列化。它们在 Kafka 主题中的原始字节格式与 Kafka Connect 使用的内部结构化数据表示法之间进行转换。

• 对于接收器连接器，转换器会将数据从主题的有线格式反序列化为 Kafka Connect 内部数据格式，连接器使用该格式写入目标系统。

• 对于源连接器，转换器会将数据从 Kafka Connect 内部数据格式序列化为 Kafka 主题的指定网络格式。

转换器可确保连接器以与外部系统兼容的格式读取或写入 Kafka 记录。

配置连接器时，请设置以下属性：

• 键转换器 (key.converter)：在序列化和反序列化 Kafka 记录键时要使用的转换器。

• 值转换器 (value.converter)：在序列化和反序列化 Kafka 记录值时要使用的转换器。

如果您未指定转换器，则默认转换器类型为 org.apache.kafka.connect.converters.ByteArrayConverter，它会以原始字节格式传递数据。

支持的转换器

Managed Service for Apache Kafka 支持以下内置转换器：

转换器的选择取决于连接器类型以及您在 Kafka 中存储的数据。如需了解详情，请参阅特定连接器的文档。

Tasks

连接器通过创建一项或多项并行运行的任务来传输数据。如需设置连接器创建的任务数量上限，请设置连接器的 tasks.max 配置属性。连接器创建的任务数可能少于此值。

增加 tasks.max 的值可以提高吞吐量，但也会增加资源消耗（CPU 和内存）。最佳值取决于工作负载以及分配给 Connect 集群工作器的资源。对于接收器连接器，Kafka 主题分区的数量也会影响并行性。

任务重启政策

您可以设置连接器的任务重启政策，该政策决定了发生故障时的行为。连接器支持以下政策：

• 从不重启。连接器不会重启失败的任务。此政策是默认行为。这在调试或出错后需要手动干预的情况下非常有用。

• 使用指数退避算法重启。连接器会在延迟（称为退避期）后重启失败的任务。延迟时间会随着每次后续失败而呈指数级增加。建议将此政策用于大多数生产工作负载。

  如果您使用指数退避算法政策，还需设置最小和最大退避算法的值。最小退避时间应大于 60 秒，最大退避时间应小于 7200 秒。

转换和谓词

Managed Service for Apache Kafka 支持默认的 Kafka Connect 转换和谓词。

借助转换，您可以在将单条消息发送到 Managed Service for Apache Kafka（对于来源连接器）或外部系统（对于接收器连接器）之前对其进行修改。您可以使用转换来遮盖敏感数据、添加时间戳或重命名字段。

借助谓词，您可以根据特定条件过滤数据，从而根据消息属性确定转换适用于哪些消息。

例如，如需将接收器连接器配置为忽略包含 DoNotProcess 标头键的消息，请添加以下配置：

transforms=dropMessage
transforms.dropMessage.type=org.apache.kafka.connect.transforms.Filter
transforms.dropMessage.predicate=hasKey
predicates=hasKey
predicates.hasKey.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.hasKey.name=DoNotProcess

此配置会执行以下操作：

1. 配置一个名为 hasKey 且类型为 org.apache.kafka.connect.transforms.predicates.HasHeaderKey 的谓词。此谓词会匹配包含键为 DoNotProcess 的标头的所有消息。

2. 配置名为 dropMessage 且类型为 org.apache.kafka.connect.transforms.Filter 的转换。此转换会丢弃与配置的谓词匹配的所有消息。

3. 将转换关联到谓词 hasKey。这可确保转换仅丢弃具有 DoNotProcess 标头键的消息。

接下来怎么做？

• Kafka Connect 概览

• 创建 Connect 集群