一文看懂大数据集结神器Kafka Connect

文 | 辛丽华

数据的每一个字节都有故事，他们预示着未来。

在数据驱动的企业中，数据迁移几乎变得和数据一样重要。

数据的速度和敏捷性决定了数据的价值。

Kafka 的前世今生

众所周知，Kafka是一种高吞吐量的分布式发布订阅消息系统，是不同系统间消息传输的纽带，是一著名的消息处理中间件。

Kafka最初产生于LinkedIn，用于支撑LinkedIn的activity stream data 和operational metrics分析，被誉为LinkedIn的“中枢神经系统”。2011年完成Apache开源 ，2012年10月完成孵化，2014年ApacheKafka中三位核心人员Jay Kreps，NehaNarkhede和 Jun Rao联合成立Confluent公司，致力于为企业提供实时数处理服务解决方案。

 （图片来自Kafka官网）

Kafka Connect 的应运而生

许多公司都采用了Apache Kafka为他们的数据管道，包括LinkedIn、Netflix和Airbnb。Kafka处理高吞吐量实时数据的能力使得它非常适合于解决数据集成问题，成为所有数据的公共缓冲区，并弥补streaming 和批处理系统之间的差距。

然而，基于Kafka构建数据管道是具有挑战性的，因为它结合各种各样的工具才能从不同的数据系统中收集数据。一个工具从数据库中导入更新，另一个导入日志，再有一个导出到HDFS等等。这就导致了构建一个数据管道需要大量的工程设计并带来了高昂的运维开销。另外，其中的一些工具并不胜任这项工作，数据集成工具生态系统的零碎性导致了创造性但误导性的解决方案，比如滥用流处理框架以实现数据集成的目的。

Kafka Connect提供了一种整合Kafka和其他系统数据的框架，以Kafka为支撑，为Kafka和其它系统创建并管理可扩展的、可信赖的流数据提供了模型。它可以通过Connect简单快捷的将大数据从Kafka中导入导出，数据范围涵盖关系型数据库，日志和度量数据，Hadoop和数据仓库，NoSQL数据存储，搜索索引等等。还有一点也很重要，相对于Consumer和Producer，KafkaConnect省掉了更多的开发工作，尤其是编码部分，这使得应用开发人员更容易上手使用。

Kafka Connect 的看家本领

Kafka Connect与用户交互的核心概念是Connector。Connector是一个逻辑作业，负责管理Kafka和另一个系统之间的数据复制，每个Connector可以实例化一组实际复制数据的task。 通过简单配置将单个job分解为多个task，Kafka Connect提供了内置的并行支持和可扩展的数据复制。Connector和tasks 是工作的逻辑单元，必须在进程中执行。Kafka Connect将这些进程称为workers。

Kafka Connect有两个核心概念：Source和Sink。 Source负责导入数据到Kafka，Sink负责从Kafka导出数据，它们都被称为Connector。

 （图片来自Confluent官网）

Kafka Connect 提供了以下特性：

• Kafka连接器通用框架：规范化其他数据系统与Kafka的集成，简化了连接器开发，部署和管理

• 分布式和单机模式：上至大型的 持整个organization的集中管理服务，下至开发、测试和小规模的生产部署

• REST接口：使用REST API提交和管理Connector

• 自动offset管理：只需从Connector获取一些信息，Kafka Connect就可以自动管理offset提交，因此连接器开发人员不需要担心这个容易出错的部分

• 分布式和可扩展性：Kafka Connect基于现有的组管理协议。可以添加更多的worker来扩展Kafka Connect群集。

• 流/批量集成：利用Kafka的现有功能，Kafka Connect是桥接流媒体和批数据系统的理想解决方案

Kafka Connect 的实战导航

workers有两种部署方式：Standalone和Distributed。

Standalone 模式

单机模式下，所有的工作运行在同一进程中完成。

启动方式：

第一个参数是worker的配置，序列化格式以及提交offset的频率等设置，紧接着的参数是Connector配置文件。可以配置多个Connector，多个Connector运行在在同一进程的不同线程中。

worker必须配置：

• bootstrap.servers：Kafka服务器列表

• key.converter：Key转换器类。常见格式的示例包括JSON和Avro。

• value.converter：Value的转换器类。常见格式的示例包括JSON和Avro。

另外，Standalone 模式还有一项重要配置：

• offset.storage.file.filename - 存储offset的文件

以自带的file的sink和source为例：

test.txt文件内容：

启动source：

启动sink：

sink控制台会输出文件内容：

继续往文件中追加内容：

Distributed模式

分布式模式能够自动平衡worker，允许动态扩展/收缩，并在活动task以及配置和offset数据提交中提供容错机制。分布式模式下worker运行的一个基本场景：

（图片来自Confluent官网）

启动方式：

和单机模式的区别在于启动的类和配置参数，这些参数改变了KafkaConnect进程如何决定存储配置，如何分配工作，以及在哪里存储偏offset和task状态。 在分布式模式下，Kafka Connect在Kafka topic存储offset，配置和task状态。 建议手动创建offset，配置和状态的topic，这样方便控制分区数和复制因子。 如果在启动Kafka Connect时尚未创建主题，那么将使用默认分区数和复制因子自动创建主题，如果使用自动创建功能，task启动时会无法获取到topic信息，task会无法正常工作。

另外的一些重要配置：

• group.id（默认topic：connect-cluster）：Connect集群组的唯一名称；请注意，不能与消费者组ID冲突
  

• config.storage.topic（默认topic：connect-configs）：用于存储Connector和task配置的topic；请注意，这应该是单分区，多副本，压缩的主题。 最好手工创建

• offset.storage.topic（默认topic：connect-offsets）：用于存储offset的topic；该topic应该有很多分区，多副本，使用压缩

• status.storage.topic（默认 connect-status）：用于存储状态的topic；此topic可以有多个分区，应该使用多副本，压缩机制

分布式的Connector需要通过rest API来维护。本文通过firefox的rest插件搞定：

以自带的file的sink和source为例：

启动：

添加消息头：

增加sink和sourceConnector：

Sink：

Source：

REST API还另外提供了查询/修改/删除等维护接口，可参照官网说明，在此不再赘述。

此时消费test topic：

通过配置禁用schema，可以清掉topic中的schema信息：

近期，Kafka还提供了一项很实用的功能，可以通过配置transformations 进行轻量级的消息即时修改， 它们可以方便数据传送和事件路由，Kafka Connect的功能越来越强大。

Kafka Connect 的前景展望

目前Kafka Connect已经支持包括HDFS、JDBC、Elasticsearch、HBsase等 五六十种Connector，并仍然在发展壮大中。Kafka以稳定、健壮、高效的姿态在众多大数据组件中以时间和效率拔得头筹，Kafka Connect功能也日益引人注目。

正如本文开始所描述，数据的速度和敏捷性决定了数据的价值，价值就是市场导向。

邬贺铨院士说：

这本书填补了大数据技术与实际应用系统性结合类书籍的空白。

当当和京东说：

这本书放在自营书店卖。

中兴大数据平台DAP团队说：这本书是我们的诚意之作。