Kafka Connect API

Kafka Connect是一个用于将数据流输入和输出Kafka的框架。

Kafka Connect基本概念

Kafka Connect实际上是Kafka流式计算的一部分
Kafka Connect主要用来与其他中间件建立流式通道
Kafka Connect支持流式和批处理集成

Kafka connect 架构图

Kafka Connect关键词

Connectors：通过管理task来协调数据流的高级抽象
Tasks：如何将数据复制到Kafka或从Kafka复制数据的实现
Workers：执行Connector和Task的运行进程
Converters：用于在Connect和外部系统发送或接收数据之间转换数据的代码
Transforms：更改由连接器生成或发送到连接器的每个消息的简单逻辑

Connectors

Kafka Connect中的connector定义了数据应该从哪里复制到哪里。connector实例是一种逻辑作业，负责管理Kafka与另一个系统之间的数据复制。

我们在大多数情况下都是使用一些平台提供的现成的connector。但是，也可以从头编写一个新的connector插件。在高层次上，希望编写新连接器插件的开发人员遵循以下工作流：

Task

Task是Connect数据模型中的主要处理数据的角色，也就是真正干活的。每个connector实例协调一组实际复制数据的task。通过允许connector将单个作业分解为多个task，Kafka Connect提供了内置的对并行性和可伸缩数据复制的支持，只需很少的配置。

这些任务没有存储任何状态。任务状态存储在Kafka中的特殊主题config.storage.topic和status.storage.topic中。因此，可以在任何时候启动、停止或重新启动任务，以提供弹性的、可伸缩的数据管道

Workers

Workers是负责管理和执行connector和task的，Workers有两种模式，Standalone（单机）和Distributed（分布式）。
Standalone Workers：

Standalone模式是最简单的模式，用单一进程负责执行所有connector和task

Distributed Workers：

Distributed模式为Kafka
Connect提供了可扩展性和自动容错能力。在分布式模式下，你可以使用相同的组启动许多worker进程。它们自动协调以跨所有可用的worker调度connector和task的执行。
如果你添加一个worker、关闭一个worker或某个worker意外失败，那么其余的worker将检测到这一点，并自动协调，在可用的worker集重新分发connector和task。

Task Rebalance

当connector首次提交到集群时，workers会重新平衡集群中的所有connector及其tasks，以便每个worker的工作量大致相同。当connector增加或减少它们所需的task数量，或者更改connector的配置时，也会使用相同的重新平衡过程。

当一个worker失败时，task在活动的worker之间重新平衡。当一个task失败时，不会触发再平衡，因为task失败被认为是一个例外情况。因此，失败的task不会被框架自动重新启动，应该通过REST API重新启动。

Converters

在向Kafka写入或从Kafka读取数据时，Converter是使Kafka Connect支持特定数据格式所必需的。task使用Converters将数据格式从字节转换为连接内部数据格式，反之亦然。并且Converter与Connector本身是解耦的，以便在Connector之间自然地重用Converter。

默认提供以下Converters：

AvroConverter（建议）：与Schema Registry一起使用 JsonConverter：适合结构数据
StringConverter：简单的字符串格式 ByteArrayConverter：提供不进行转换的“传递”选项
AvroConverter处理数据的流程图：

Transforms

Connector可以配置Transforms，以便对单个消息进行简单且轻量的修改。这对于小数据的调整和事件路由十分方便，且可以在connector配置中将多个Transforms连接在一起。然而，应用于多个消息的更复杂的Transforms最好使用KSQL和Kafka Stream来实现。

Transforms是一个简单的函数，输入一条记录，并输出一条修改过的记录。Kafka Connect提供许多Transforms，它们都执行简单但有用的修改。可以使用自己的逻辑定制实现转换接口，将它们打包为Kafka Connect插件，将它们与connector一起使用。

当Transforms与Source Connector一起使用时，Kafka Connect通过第一个Transforms传递connector生成的每条源记录，第一个Transforms对其进行修改并输出一个新的源记录。将更新后的源记录传递到链中的下一个Transforms，该Transforms再生成一个新的修改后的源记录。最后更新的源记录会被转换为二进制格式写入到Kafka。Transforms也可以与Sink Connector一起使用。

以下为Confluent平台提供的Transforms：https://docs.confluent.io/current/connect/transforms/index.html

Kakfa Connect环境准备

前面已经铺垫了Kakfa Connect的基本概念，接下来用一个简单的例子演示一下Kakfa Connect的使用方式，以便对其作用有一个直观的了解。

在演示Kakfa Connect的使用之前我们需要先做一些准备，因为依赖一些额外的集成。例如在本文中使用MySQL作为数据源的输入和输出，所以首先得在MySQL中创建两张表（作为Data Source和Data Sink）。建表SQL如下：

CREATE TABLE `users_input` (`uid` int(11) NOT NULL AUTO_INCREMENT,`name` varchar(20) NOT NULL,`age` int(11) NOT NULL,PRIMARY KEY (`uid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;CREATE TABLE `users_output` (`uid` int(11) NOT NULL AUTO_INCREMENT,`name` varchar(20) NOT NULL,`age` int(11) NOT NULL,PRIMARY KEY (`uid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

接下来就是考虑怎么实现Kafka Connect了，前面有提到过Kafka Connect中的connector定义了数据应该从哪里复制到哪里。connector实例是一种逻辑作业，负责管理Kafka与另一个系统之间的数据复制。

因此，如果要自己实现一个Connect的话还是稍微有些复杂的，好在Confluent平台有些现成的Connect。例如Confluent平台就有JDBC的Connect，下载地址如下：
https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc