07 Confluent_Kafka权威指南 第七章： 构建数据管道

CHAPTER 7 Building Data Pipelines 构建数据管道

当人们讨论使用apache kafka构建数据管道时，他们通常会应用如下几个示例，第一个就是构建一个数据管道，Apache Kafka是其中的终点。丽日，从kafka获取数据到s3或者从Mongodb获取数据到kafka。第二个用例涉及在两个不同的系统之间构建管道。但是使用kafka做为中介。一个例子就是先从twitter使用kafka发送数据到Elasticsearch，从twitter获取数据到kafka。然后从kafka写入到Elasticsearch。 我们在0.9版本之后在Apache kafka 中增加了kafka connect。是我们看到之后再linkerdin和其他大型公司都使用了kafka。我们注意到，在将kafka集成到数据管道中的时候，每个公司都必须解决的一些特定的挑战，因此我们决定向kafka 添加AP来解决其中的一些特定的挑战。而不是每个公司都需要从头开发。 kafka为数据管道提供的主要价值是它能够在管道的各个阶段之间充当一个非常大的，可靠的缓冲区，有效地解耦管道内数据的生产者和消费者。这种解耦，结合可靠性、安全性和效率，使kafka很适合大多数数据管道。

将数据集成到上下文

一些组织认为kafka是数据管道的终点。他们关注的问题是，我如何从kafka弹性得到数据，这事一个值得有效提出的问题，特别是如果你需要数据保持弹性，而且它目前正在kafka中。我们将寻找方法来解决这一点。但是我们将通过一个更大的背景下看kafka的使用来开始讨论。这个背景包括至少两个不是kafka本身的终点。我们鼓励任何面临数据集成问题的人从更大的角度考虑问题，而不是只关注数据本身，关注于短期集成将导致复杂且维护成本高安的数据集成混乱。 在本章中，我们将讨论在构建数据管道时需要考虑的一些常见问题。这些挑战不是kakfa特有的，而是一般的数据集成问题，尽管如此，我们将展示为什么kafka非常适合数据集成的用例场景，以及它是如何解决这些挑战的。 我们将讨论kafka connect API与普通的生产者和消费者客户端有何不同，以及何时应该使用每种类型的客户端，然后我们将跳转到kafka connect的一些细节。虽然关于kafka connect的完整讨论超出了本章的范围，但是我们将展示一些基本的用法和例子来让你开始学习，并给你更多的指导。最后我们将讨论其他的数据系统如何与kafka集成。

Considerations When Building Data Pipelines 构建数据管道时需要考虑的内容

虽然我们不在此讨论构建数据管道的所有细节，但是我们想强调在涉及旨在集成多个系统的软件架构时需要考虑的一些重要的事情。

Timeliness 时效性

一些系统希望将他们的数据每条一次大批量的到达，其他人则希望数据叜生成几毫秒后就到达。大多数数据管道介于这两个极端情况之间。良好的数据集成系统可以支持不同管道的不同及时性需求，还可以简化不同的时间表之间的迁移，因为业务需求可能会发生变化。Kafka是一个具有可伸缩性的可靠存储的流平台，合一用于支持从近实时管道到每小时批量的任何数据，生产者可以写消息给消费者，还可以根据需要读取和发布最新消息。消费者可以批量工作，每小时运行一次，连接到kafka并读取前一小时累计的消息。 在这种情况下，看代kafka的一个有用的方法是，它充当了一个巨大的缓冲区，解耦了生产者和消费者之间的时间敏感性需求。生产者可以在消费者处理成批消息实时写入消息，反之亦然。这也使得应用背压，kafka本身对生产者施加压背压（通过在需要时延迟acks）变得微不足道。因为消费率完全由消费者者驱动。

Reliability 可靠性

我们希望避免单点故障，并允许从各种故障事件中快速恢复。数据管道通常时数据到达业务关键系统的途径。超过几秒的故障会造成巨大的破坏。特别是当实时性需求接近毫秒时。另外一个重要考虑是交付保证，有些系统可以承受数据丢失，但是大多数时候需要至少传递一次，这意味着来自源系统的每个事件都会到达目的地。但是重试会导致重复。通常甚至需要exactly-once 来自源系统的每个事件都将到达目的地，不存在丢失或者重复的可能性。 我们在第6章中深入的讨论了kafka的可用性和可靠性保证。正如我们所讨论的，当kafka具有事务模型或者唯一键的外部数据存储到一起时。它可以实现exactly-once交付。由于许多数据节点都是为exactily-once提供正确语义的数据存储，基于kafka的管道通常可以实现exactily-once。值得一提的时，kafka的connect api通过提供与外部系统集成的api。使connect更容易的构建端到端的exactily-once管道。

High and Varying Throughput 高而多变的吞吐量

我们正在构建的数据管道应该能够扩展到非常高的吞吐量。这在线到的数据系统中是非常有必要的。更重要的是，如果吞吐量突然增加，他们应该能够适应。 有了kafka为生产者和消费者之间的缓冲，我们不再需要将消费者和生产者的产能耦合。我们不再需要实现负载的背压机制，因为如果我们生产者中的产量都超过了消费者的消费能力，数据就会在kafak中累计。直到消费者赶上来。kafka通过独立添加的消费者或者生产者来扩展的能力允许我们动态独立地扩展管道的任何一端，以匹配不断变化的需求。 kafka是一个高吞吐量的分布式系统，即使在中等规模的集群上也能每秒处理数百M字节，因此我们不需要担心管道会随着需求的增长而无法扩展。此外，kafka connect API关注的并行化工作，而不仅仅是扩展。在下面的部分中，我们将描述该平台如何允许数据源和接收在多个执行线程之间分隔工作。并使用可用的CPU资源。即使只有一台服务器。 kafka还支持集中数据类型，允许用户和管理员在吞吐量需求增加时降低网络和存储资源的使用。

Data Formats 数据格式

数据管道一个最重要的需求就是实现了不同节点间的数据格式和数据类型。不同的数据库和其他存储系统所支持的数据类型各不相同。你可能将使用kafka中的avro格式将xml数据加载到kafka中。然后将数据转换为json存储到elasticsearch。最后写入HDFS和S3时转换为csv。 当涉及到数据格式的时候，kafak本身和connect api是完全不可知的。正如我们在前几章所看到的，生产者和消费者可以使用任何序列化器以任何适合你的格式表示数据。kafka connecct有自己的内存对象，包括数据类型和模式。但是我们很快就会讨论，它允许可插接的转换器以任何格式存储这些记录。这意味着无论你为kafka使用那种数据格式，他都不会限制你对连接器的选择。 许多源和接收器都有一个模式，我们可以从数据源读取带有数据的模式，存储它，并使用它来验证兼容性。甚至sink数据库中的模式。一个典型的例子就是从mysql到hive的数据管道。如果有人在mysql中增加了一列，一个好的管道将确保该列被添加到hive。因我正在加载新的数据格式的数据到hive。 此外，当从kafka写入数据到外部系统的时候，sink连接器将负责将数据写入到外部系统所支持的格式中。一些连接器选择使用这种格式配置，例如，kdfs连接器允许在avro和parquet上做出选择。 仅仅支持不同类型的数据是不够的，通用的数据集成框架还应该处理各种源和目标数据库之间的行为差异。例如，syslog是一个推送数据的源，而关系数据库需要框架来写入数据。hdfs只允许追加数据，我们只能向它写入数据，而大多数系统允许我们追加数据和更新现有数据。

Transformations 转换

转换比其他的需求特性更具有争议，通常由两种构建数据转换的学派，ETL和ELT。ETL即使Extract-Transform-Load 意味着数据管道负责在数据通过时对其进行修改。它可以节省时间和存储空间。因为你不需要存储、修改和再次存储数据。根据转换的不同，这种好处是实在的，但是有时候将计算和存储的负担转移到了管道本身，着可能是所希望的，也可能不是。这种方法的主要缺点就是，对管道中的数据进行转换时束缚了希望进一步处理管道中数据的人员的手脚。如果在mongodb和mysql之间建立管道的人决定过滤某些事件或从激励中删除字段，那么所有访问mysql数据的用户和应用程序只能访问部分数据。如果它需要访问丢失的字段，则需要重新构建管道，并且需要重新处理历史数据。 ELT表示Extract-Load-Transform ，这意味着数据管道只能进行很少的转换，主要是数据类型的转换。其目标是确保到达的目标的数据尽可能的与源相似。这也被称为高保真管道或者数据湖架构。在这些系统中，目标系统收集原始数据，所有需要的处理都在目标系统上完成。这样做的好处就是，系统为目标系统的用户提供了大的灵活性。因为他们可以访问所有的数据。这些系统往往也更加容易排除故障。因为所有的数据存储都在要给系统。而不是在管道和其他程序之间分隔。缺点是转换占用了目标系统上CPU和存储资源。在某些情况下。这些系统资源是昂贵的，并且有很大的可能将这些计算移除系统。

Security 安全性

安全性一直是一个值得关注的问题，在数据管道方面，主要的安全性问题有：

• 我们能确保通过管道数据是加密的吗？这主要是在跨数据中心边接的数据管道需要考虑的问题。
• 谁可以对管道进行修改。
• 如果数据管道需要从访问控制的位置读写，他能正确的进行身份验证吗？
• kafka允许加密数据发送，支持kafka从数据来源到管道和从kafka到写入的数据节点。它还支持身份验证通过sasl和授权，可以肯定的是，如果topic包含了敏感信息，则不能通过未授权的人将其导入安全性较低的系统中。kafka还提供了一个审计日志来跟踪未授权的访问和已授权的访问，通过一些额外的变慢，还可以跟踪每个topic中的事件来自何处以及谁修改了他们，因此可以为每个记录提供整个数据血缘。

Failure Handling 故障处理

假设我们所有的数据在任何时候都是安全的，这种想法是危险的。提前计划故障处理很重要。我们能阻止错误的记录进入数据管道吗？我们能从无法解析的记录中恢复吗 ？坏记录能被修复，并重新处理吗？如果坏的事件看起来与正常的事件完全一样，而你知识在几天后才发现问题，哪应该怎么办？ 因为kafka长时间存储所有消息。所以在需要的时候可以从错误中恢复。

Coupling and Agility 耦合和敏捷

数据管道最重要的目标之一是解耦数据源和数据目标。偶然耦合的发生有多种方式：

Ad-hoc pipelines 特定管道

一些公司最终会为他们想要连接的程序构建一个定制的数据管道。例如，他们使用logstash将日志转储到elasticsearch。通过flume将数据转储到hdfs。GoldenGate将oracel的数据转储到hdfs。informatica从mysql和xml获取数据写入到oracle等等。这将数据管道紧密地连接到了特定的端点，并创建了一堆集成的点，需要带昂的工作来部署，维护和监视。这意味着，该公司采用的每一个新系统都将奖励更多的管道，增加采用新技术的成本，并抑制创新。

Loss of metadata 元数据丢失

如果数据管道不保存模式和元数据，并且不允许模式演化，那么最终将导致源产生的数据的软件与在目标使用的数据软件紧密耦合。如果没有模式信息，两个软件都要包含关于如何解析和解释数据信息。如果数据从oracle到hdfs，并且dba在oracle中添加了一个新字段，而且没有保存模式信息并允许模式演化，那么要么每个重从hdfs读取的数据的应用程序都会崩溃，要么所有的开发人员都需要同时升级他们的应用程序。这两种选择都不是敏捷的，通过支持管道中的模式演化，每个团队都可以进行修改。他们的应用程序则按照自己的进度进行，而不会担心这些事情会打乱进度。

Extreme processing 极端处理

正如我们在讨论数据转换时提到，一些数据处理是数据管道固有的，毕竟，我们是在不同的系统之间移动数据，在这些系统中，不同的数据格式是有意义的，并且支持不同的用例。然而，太多的处理将所有下游系统与构建在管道时所做的决策联系在一起，关于保留哪些字段，如何聚合数据等决策。这通常会导致随着下游的应用程序需求的变化而对管道进行不断的更改，这不是敏捷的，有效和安全的。更敏捷的方法保存尽可能多的原始数据，让下游的应用程序自行决定数据处理和聚合。

When to Use Kafka Connect Versus Producer and

Consumer 何时使用连接器（在生产者和消费者上） 当你发送消息给kafka或者从kafka读取消息时，你可以选择传统的生产者和消费者客户端。或者使用connect API，在我们开始深入了解connect的细节之前，有必要问问自己，什么时候，该使用哪个？ 正如我们所见，kafak客户端时嵌入在你自己的应用程序中的客户端。它允许你的应用程序写入数据到kafka或者从kafka中读取数据。当你可以修改你想要连接的应用程序的代码时，或者当你想要将数据推入kafka或者从kafka提取数据时，请使用kafka客户端。 你将使用connect将kafka连接到你没有编写且你不打算修改其代码的数据存储中。connect将用于从外部存储中拉取数据到kafka或者将数据从kafka推送到外部存储中。对于已经存在的连接器和数据存储。非开发人员使用connect，他们只需要配置连接器即可。 如果需要将kafka连接到数据存储，而连接器还不存在，你可以选择使用kafak客户端，或者connect API编写应用程序。推荐使用connect，因为它提供了开箱即用的特性。如配置管理、偏移存储，并行化、错误处理，对不同数据类型支持以及标准的管理REST API。 编写一个连接的小的应用程序将kafka用于数据存储听起来很简单。但是你需要将处理许多数据类型的配置和小细节。这使得这项任务不那么简单。kafka connect将为你处理大部分工作。是你能够集中精力将数据传输到外部存储区和从外部存储区传入到外部存储区。

Kafka Connect kafka连接器

kafka connect 时kafka的一部分，它提供了一种弹性且可靠的方式在kafka和其他数据存储中移动数据。它提供了API和运行时开发和运行连接器的插件，kafka connect 执行的负责移动数据的数据库。kafka connect做为一个工作进程的方式集群运行。你将在worker上安装连接器的插件，然后使用REST API来配置和管理连接器，连接器使用特定的配置运行。连接器启动额外的任务，以并行地移动大量数据，并更有效地使用工作节点上的可用资源。源连接器任务只需要从源系统读取数据，并将连接数据对象提供给工作进程。接收连接器的任务从工作程序中获得连接器的数据对象，ing负责将他们写入目标数据系统。kafka connect使用转换器来支持kafka中存储的不同格式的数据对象。json格式支持是kafka的一部分。Confluent的模式注册中心提供了avro的转换器。这运行用户选择数据存储格式与他们的连接器无关。 本章不可能对kafka connect的各种细节进行讨论，这就足够写一本书。然而，我们将给出kafka connect的概述以及如何使用他们，并之处其参考的资源配置。

Running Connect 运行连接器

kafka 的connect是与apache kafka一起发布的，所以没有必要单独安装它，对于生产使用，特别是计划使用connect移动大量数据或运行多个连接器时，应该在单独的服务器上运行connect。在这种情况下，所有的及其上安装apache kafka，并在一些服务商启动broker，在其他服务器上启动connect。 启动一个连接worker 与启动一个broker非常相似，你用一个属性文件调用启动脚本：

bin/connect-distributed.sh config/connect-distributed.properties

有几个连接器的关键配置：

• bootstrap.servers connect 将与之合作的kafka的broker列表，连接器将其数据传输到此或者从这些broker中读取数据，你不需要指定集群的每个broker但是建议至少不少于3个broker。
• group.id 具有相同组ID的所有worker都属于同一个connect集群。在集群上启动的connect将在任何worker上运行，它的任务也是如此。
• key.converter and value.converter 连接器可以处理多种数据格式存储在kafka，这两种配置将为存储在kafka中的消息的key和value部分设置了转换器。默认是使用apache kafka中包含的JSON converter的json格式，也可以设置为Avro Converter,它是Confluent 模式注册表的一部分。 有些转换器包含特定于转换器的配置参数，例如，JSON消息可以包含模式，也可以不包含模式。要支持这两者，可以设置 key.converter.schema.enable=true或者 false ，Avro也可以使用相同的这些配置，需要注意的是 Avro需要配置注册中心的位置：key.converter.schema.registry.url 和 value.converter.schema.registry.url 。 rest.host.name and rest.port 连接器通常是通过kafka connect的REST API进行配置和监控。你能通过REST API进行特定的配置。 一旦这些workers启动，你有一个集群，以确保它是启动和运行状态,可以通过REST API进行查询：

gwen$ curl http://localhost:8083/
{"version":"0.10.1.0-SNAPSHOT","commit":"561f45d747cd2a8c"}

访问基本的REST API 应该返回你正在运行的当前版本。我们正在运行的版本是 kafka 0.10.1.0,我们还可以检查哪些连接器插件可用：

gwen$ curl http://localhost:8083/connector-plugins
[{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector"},
{"class":"org.apache.kafka.connect.file.FileStreamSinkConnector"}]

我们运行的是普通的apache kafka ，因此唯一可用的连接器插件是文件源和文件接收器。 让我看看如何配置和使用这些连接器，然后我们将深入一些高级的示例，这些示例需要设置连接器的外部数据系统。

Standalone Mode 独立运行模式

注意，kafka connect也有一个独立模式，它与分布式模式类似，只运行bin/connect-stadalone.sh 你还可以通过命令行传递连接器的配置文件，而不是通过rest API。在此模式下，所有的连接器和任务都运行在一个独立的worker上。在独立模式下使用connect进行开发和故障诊断，以及在连接器和任务需要的运行在特定机器上的情况下，通常更容易。例如：syslog.connector 监听的端口，因此你需要知道它允许在哪个机器上。

Connector Example: File Source and File Sink 连接器示例：文件源和文件接收器

本例将使用APache的文件连接器和j属于kafka的json转换器。接下来，确保你又zookeeper和kafka启动并允许。首先，让我们允许一个分布式连接器的worker。在实际的生产环境中，你需要至少允许两到三个这也的系统来提供高可用性。在这个例子中，我们只启动一个：

bin/connect-distributed.sh config/connect-distributed.properties &

现在启动文件源，做为示例，我们将配置它来读取apache kafka的配置文件，kafka的基本配置到kafka的topic中。

echo '{"name":"load-kafka-config", "config":{"connector.class":"FileStream-
Source","file":"config/server.properties","topic":"kafka-config-topic"}}' |
curl -X POST -d @- http://localhost:8083/connectors --header "content-
Type:application/json"
{"name":"load-kafka-config","config":{"connector.class":"FileStream-
Source","file":"config/server.properties","topic":"kafka-configtopic","
name":"load-kafka-config"},"tasks":[]}

为了创建连接器，我们编写了一个JSON，其中包含连接器的名称 load-kafka-config 和连接器配置映射，其中包含连接器类，要加载的文件和要加载的文件的toppic。 让我们使用kafka控制台的消费者来检查我们以及将配置加载到一个topic中。

gwen$ bin/kafka-console-consumer.sh --new --bootstrap-server=localhost:9092 --
topic kafka-config-topic --from-beginning

如果一切顺利，你将能看到如下内容：

{"schema":{"type":"string","optional":false},"payload":"# Licensed to the
Apache Software Foundation (ASF) under one or more"}
<more stuff here>
{"schema":{"type":"string","optional":false},"payload":"#############################
Server Basics
#############################"}
{"schema":{"type":"string","optional":false},"payload":""}
{"schema":{"type":"string","optional":false},"payload":"# The id of the broker.
This must be set to a unique integer for each broker."}
{"schema":{"type":"string","optional":false},"payload":"broker.id=0"}
{"schema":{"type":"string","optional":false},"payload":""}
<more stuff here>

这事配置/服务器的属性文件内容，因为它被转换为我们连接器的JSON并一行一行地放在kafka-config-topic中。注意，默认情况下，JSON专户去的每个记录中放置一个模式。在这个特定的例子中，模式非常简单。只有一个名为payload的列，类型为String，它包含文件中每一个记录的一行。 现在让我们使用文件的接收转换器将该topic的内容转储到一个文件中，结果文件应该与原始服务器完全相同。属性文件因为JSON转换器将json记录转换为简单的文本行。

echo '{"name":"dump-kafka-config", "config":
{"connector.class":"FileStreamSink","file":"copy-of-serverproperties","
topics":"kafka-config-topic"}}' | curl -X POST -d @- http://localhost:
8083/connectors --header "content-Type:application/json"
{"name":"dump-kafka-config","config":
{"connector.class":"FileStreamSink","file":"copy-of-serverproperties","
topics":"kafka-config-topic","name":"dump-kafka-config"},"tasks":
[]}

请注意来自源配置的更改，我们使用的类现在是FileStreamSink 而不是FileStreamSource。我们仍然有一个file属性，但是现在它引用的是目标文件而不是记录的源。并且指定的topic而不是指定的主题。注意它的多元性，你可以用接收器将多个topic写入一个文件，而源只允许写入一个topic。 如果一切顺利，你将有一个名为 copy-of-server-properties的文件，它与config/server.prorerties完全相同。用于填充kafka-config-topic。 要删除连接器，你可以允许如下：

curl -X DELETE http://localhost:8083/connectors/dump-kafka-config

如果在删除连接器后查看connect work日志，应该会看代所有其他连接器都在重启他们的任务。他们重启是为了在删除连接器之后worker和剩余任务之间的平衡，确保每个worker有相同的工作负载。

Connector Example: MySQL to Elasticsearch 连接器示例，mysql到Elasticsearch

现在我们有一个简单的示例，让我们做一些更有用的事情，让我们取一个mysql的表，将其转换为kafka的topic，然后将其加载到elasticsearch并索引其内容。 我们在一台macbook上允许测试，安装了Mysql和Elasticsearch，我们只需要运行：

brew install mysql
brew install elasticsearch

下一步是确保你有连接器，如果你在运行confluent，如果是开源的， 你应该将连接器做为平台的一部分安装好。否则，你可以直接从github上构建：

1. Go to https://github.com/confluentinc/kafka-connect-elasticsearch
2. Clone the repository
3. Run mvn install to build the project
4. Repeat with the JDBC connector

现在你将在目标目录下创建的jar文件复制到kafka connect的类的路径中：

gwen$ mkdir libs
gwen$ cp ../kafka-connect-jdbc/target/kafka-connect-jdbc-3.1.0-SNAPSHOT.jar
libs/
gwen$ cp ../kafka-connect-elasticsearch/target/kafka-connectelasticsearch-
3.2.0-SNAPSHOT-package/share/java/kafka-connect-elasticsearch/*
libs/

如果kafka connect 的worker还没有运行，请确保他们启动，并检查列出的新的连接器插件：

gwen$ bin/connect-distributed.sh config/connect-distributed.properties &
gwen$ curl http://localhost:8083/connector-plugins
[{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector"},
{"class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector"},
{"class":"org.apache.kafka.connect.file.FileStreamSinkConnector"},
{"class":"io.confluent.connect.jdbc.JdbcSourceConnector"}]

我们可以看代，现在我们的connect集群中有了额外的连接器插件。JDBC源需要一个Mysql驱动程序才能使用mysql。我们从oralce的官网下载了mysql的驱动，解压，并复制 mysql-connector-java-5.1.4-bin.jar复制到lib目录。 下一步是在mysql中创建一个表，我们可以使用我们的kafka JDBC连接器：

gwen$ mysql.server restart
mysql> create database test;
Query OK, 1 row affected (0.00 sec)
mysql> use test;
Database changed
mysql> create table login (username varchar(30), login_time datetime);
Query OK, 0 rows affected (0.02 sec)
mysql> insert into login values ('gwenshap', now());
Query OK, 1 row affected (0.01 sec)
mysql> insert into login values ('tpalino', now());
Query OK, 1 row affected (0.00 sec)
mysql> commit;
Query OK, 0 rows affected (0.01 sec)

我们创建了一个DataBase,并建立了一个表，插入了一些行为数据。 下一步是配置JDBC源连接器，我们可以通过差康文档找到可用的配置选项，但是我们也可以使用REST API来找到可用的配置选项：

gwen$ curl -X PUT -d "{}" localhost:8083/connector-plugins/JdbcSourceConnector/
config/validate --header "content-Type:application/json" | python -m json.tool
{
"configs": [
{
"definition": {
"default_value": "",
"dependents": [],
"display_name": "Timestamp Column Name",
"documentation": "The name of the timestamp column to use
to detect new or modified rows. This column may not be
nullable.",
"group": "Mode",
"importance": "MEDIUM",
"name": "timestamp.column.name",
"order": 3,
"required": false,
"type": "STRING",
"width": "MEDIUM"
},
<more stuff>

我们基本上要求REST API验证连接器的配置，并向他发送一个空配置，做为响应，我们得到所有可以配置json定义。我们通过管道输出，以使json更具有可读性。 记住这些信息，现在是创建和配置JDBC连接器的时候了：

echo '{"name":"mysql-login-connector", "config":{"connector.class":"JdbcSource-
Connector","connection.url":"jdbc:mysql://127.0.0.1:3306/test?
user=root","mode":"timestamp","table.whitelist":"login","validate.
non.null":false,"timestamp.column.name":"login_time","topic.prefix":"
mysql."}}' | curl -X POST -d @- http://localhost:8083/connectors --header
"content-Type:application/json"
{"name":"mysql-login-connector","config":{"connector.class":"JdbcSourceConnector","
connection.url":"jdbc:mysql://127.0.0.1:3306/test?
user=root","mode":"timestamp","table.whitelist":"login","validate.non.null":"fal
se","timestamp.column.name":"login_time","topic.prefix":"mysql.","name":"mysqllogin-
connector"},"tasks":[]}

让我们通过mysql.login配置确保他能工作：

gwen$ bin/kafka-console-consumer.sh --new --bootstrap-server=localhost:9092 --
topic mysql.login --from-beginning
<more stuff>
{"schema":{"type":"struct","fields":
[{"type":"string","optional":true,"field":"username"},
{"type":"int64","optional":true,"name":"org.apache.kafka.connect.data.Timestamp","
version":1,"field":"login_time"}],"optional":false,"name":"login"},"payload":{"
username":"gwenshap","login_time":1476423962000}}
{"schema":{"type":"struct","fields":
[{"type":"string","optional":true,"field":"username"},
{"type":"int64","optional":true,"name":"org.apache.kafka.connect.data.Timestamp","
version":1,"field":"login_time"}],"optional":false,"name":"login"},"payload":{"
username":"tpalino","login_time":1476423981000}}

如果你得到的错误说topic不存在或者你看不到任何数据，检查连接器的工作日志错误，如：

[2016-10-16 19:39:40,482] ERROR Error while starting connector mysql-loginconnector
(org.apache.kafka.connect.runtime.WorkerConnector:108)
org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: Access
denied for user 'root;'@'localhost' (using password: NO)
at io.confluent.connect.jdbc.JdbcSourceConnector.start(JdbcSourceConnector.
java:78)

需要多次尝试才能获得正确的连接字符串，其他问题可能涉及类似类路径中的驱动成员或者读取表的权限。 请注意，在连接器运行时，如果在登陆表中插入额外的行，应该会立即看到他们在mysql.login topic中的反映. 将mysql数据获取到kafka本身是有用的，但是让我们通过将数据写入到elasticsearch来使事情变得更有趣。 首先，我们启动elasticsearch，并通过访问其本地端口来验证他是UP状态。

gwen$ elasticsearch &
gwen$ curl http://localhost:9200/
{
"name" : "Hammerhead",
"cluster_name" : "elasticsearch_gwen",
"cluster_uuid" : "42D5GrxOQFebf83DYgNl-g",
"version" : {
"number" : "2.4.1",
"build_hash" : "c67dc32e24162035d18d6fe1e952c4cbcbe79d16",
"build_timestamp" : "2016-09-27T18:57:55Z",
"build_snapshot" : false,
"lucene_version" : "5.5.2"
},
"tagline" : "You Know, for Search"
}

现在启动连接器：

echo '{"name":"elastic-login-connector", "config":{"connector.class":"ElasticsearchSinkConnector","
connection.url":"http://localhost:
9200","type.name":"mysql-data","topics":"mysql.login","key.ignore":true}}' |
curl -X POST -d @- http://localhost:8083/connectors --header "content-
Type:application/json"
{"name":"elastic-login-connector","config":{"connector.class":"Elasticsearch-
SinkConnector","connection.url":"http://localhost:9200","type.name":"mysqldata","
topics":"mysql.login","key.ignore":"true","name":"elastic-loginconnector"},"
tasks":[{"connector":"elastic-login-connector","task":0}]}

这里需要我们解释的配置很少，connect.url就是我们前面配置的本地elasticsearch服务器的url，默认情况下，kafka中的每个topic都将称为一个独立的elasticsearch 索引，与topic同名。在topic中，我们需要为写入的数据定义类型，我们假设一个topic中所有消息都是相同类型，因此我们只能硬编码type.name=mysql-data。我们为elasticsearch写入的唯一topic就是mysql.login,当我们在mysql中定义表的时候，我们没有给他一个key。因此kafka中的消息的key都是空的，因为kafka中的消息缺少key，我们需要告诉elasticsearch连接器使用topic、分区id和offset做为每个消息的key。 让我们检查下mysql.login的索引是否被创建：

gwen$ curl 'localhost:9200/_cat/indices?v'
health status index pri rep docs.count docs.deleted store.size
pri.store.size
yellow open mysql.login 5 1 3 0 10.7kb
10.7kb

如果索引不存在，请在connect worker日志中查找错误，缺少配置或者database是常见的原因。如果一起顺利，我们可以在索引中搜索我们的记录。

gwen$ curl -s -X "GET" "http://localhost:9200/mysql.login/_search?pretty=true"
{
"took" : 29,
"timed_out" : false,
"_shards" : {
"total" : 5,
"successful" : 5,
"failed" : 0
},
"hits" : {
"total" : 3,
"max_score" : 1.0,
"hits" : [ {
"_index" : "mysql.login",
"_type" : "mysql-data",
"_id" : "mysql.login+0+1",
"_score" : 1.0,
"_source" : {
"username" : "tpalino",
"login_time" : 1476423981000
}
}, {
"_index" : "mysql.login",
"_type" : "mysql-data",
"_id" : "mysql.login+0+2",
"_score" : 1.0,
"_source" : {
"username" : "nnarkede",
"login_time" : 1476672246000
}
}, {
"_index" : "mysql.login",
"_type" : "mysql-data",
"_id" : "mysql.login+0+0",
"_score" : 1.0,
"_source" : {
"username" : "gwenshap",
"login_time" : 1476423962000
}
} ]
}
}

如果在mysql中向表添加新记录，他们将自动出现在mysql.login topic中和elasticsearch对应的索引中。 现在我们以及了解了如何构建和安装JDBC源和Elasticsearch的接收器，我们可以构建和使用适合我们的用例的任何一对连接器。confluent维护了我们所知的所有连接器列表，包括由公司和社区编写和支持的连接器。你可以在列表中选择你希望使用的任何连接器。从github存储库中构建它，配置它，根据文档或者RestApi 中提取配置，并在你的connect worker集群上运行它。

Build Your Own Connectors 自建连接器

连接器api是公共的，任何人都可以创建新的连接器。事实上，这就是大多数连接器成为了连接器中心-人们自建了连接器并告诉他们的情况。因此，如果你希望集成的数据库在连接器HUB中不可用，你可以自己编写并将其贡献给社区。这也其他人可以发现和使用它。 讨论所有构建连接器的细节超出了本章的范围，但是你可以在官方文档中了解它。我们还建议以现有的连接器为起点，或者可以使用maven archtype来启动，我们一直鼓励你在apache kafka社区邮件列表中寻求帮助或者展示你最新的连接器 users@kafka.apache.org.

A Deeper Look at Connect 更深层次的了解连接器

要理解connect是如何工作的，你需要理解三个基本概念，以及它们是如何交互的。正如我们前面解释的示例和演示那样，使用你需要运行的一组工作程序启动和停止连接器。我们之前没有深入讨论另外一个细节是转换器对数据的处理。转化器是将mysql行转换为json记录的组件，连接器将其写入kafka中。 让我们更深入的了解每个系统以及他们之间是如何交互的。

Connectors and tasks 连接器和任务

连接器API包括两部分：

Connectors 连接器

连接器负责三件重要的事情：

• 缺点连接器将运行多少个任务
• 决定如何在任务之间分隔数据和复制工作
• 供工作人员哪里获得任务的配置，并将其传递下去 例如，JDBC源连接器将这些连接到数据库，发送现在要复制的现有的表，然后根据这些表决定需要多少tasks，选择较低的max.tasks配置task的任务数量。一旦它决定运行多少个任务，它将为每个任务生成一个配置，使用连接器配置，如connection.url以及要为每个复制任务要分配的表list。taskConfig方法的作用是返回一个map，即我们要运行的每个任务。然后这些worker负责启动你任务，并未每个任务提供独特的配置。以便从数据库中复制唯一的表子集。注意，当你通过REST API启动连接器时，它可以在任何节点上启动，随后它启动的任务也可能在任何节点上执行。

Tasks 任务

任务负责从kafka中实际获取数据。所有任务都是通过接收worker的上下文来初始化的。源的上下文包含一个对象，该对象运行源任务存储源记录的offset（例如，在文件连接器中，offset是文件中的文章，在JDBBC源连接器中，offset可以是表的主键ID）。接收连接器的上下文包括允许连接器控制其接收的记录的方法。kafka用于应用的背压、重新尝试和在外部存储的offset以确保一次交付。在初始化任务之后，使用属性的对象启动任务，该对象包含未任务创建的连接器的配置。 一旦任务启动，源任务轮询外部系统并返回工作人员发送给kafkabroker的记录列表，接收任务通过woker从kafka接收记录，并负责将记录写入外部系统。

Workers

kafka connect的工作进程是执行连接器和任务的容器进程。他们负责处理定义连接器以及其配置的http请求，以及存储连接器配置、启动连接器及其任务传递的适当配置。如果一个工作进程停止或者崩溃，connect集群中的其他工作进程将识别（通过kafka消费者协议中的心跳机制），并将允许在该工作进程上的连接器和任务重新分配给剩余的工作进程。如果一个新的worker加入一个connect集群，其他worker会注意到这一点，并为它分配连接器和任务。以确保所有的worker之间的公平平衡。 工作人员还负责为源和接收连接器自动提交offset，并在任务抛出错误的时候处理重试。 理解worker最好的方法是认识到连接器和任务负责数据集成的移动数据部分，而worker负责的REST API、配置管理、可靠性、高可用性、扩展和负载平衡。 与传统的生产者/消费者的API相比，这种关注点分离式使用连接API的好处。校验丰富的开发人员从kafka知道写代码读取数据并将它charity到一个数据库可能需要一两天，但是如果你需要知道配置错误、REST API，监控、部署、扩展和处理故障，可能需要几个月。如果你用连接器实现数据复制，那么你的连接器可以连接到处理一堆你不需要担心的复杂操作问题的workers上。

Converters and Connect’s data model

connect API最容易出错的地方式在于连接器的数据模型和转换器。kafka的connect API包括一个数据API，它包括数据对象和描述数据的模式。例如，JDBC源从数据库中读取一个列，并根据数据库返回的列的数据类型构造一个connect模式对象。然后，它使用该模式构造一个包含数据库记录中的所有字段结构。对于每个列，我们存储的列名和列中的值，每个源连接器都做类似的事情，从源系统中读取消息并生成一对schema和value。接收连接器则恰好相反，获取schema和value并使用schema来解析值，并将他们插入目标系统。 尽管源连接器知道如何基于DATA API生成丢箱，但是任然存在一个问题，即connect workers如何在kafka中存储这些对象。这就是转化器的作用，当用户配置worker时，他们选择要使用哪个转换器在kafka中存储数据。目前可以选择的式acro，JSON或者字符串。JSON专户去可以配置为在结果激励中包含模式或者不包含模式，因此我们可以同时支持结构化和半结构化的数据。连接器返回数据 API的记录给worker,然后worker使用配置的转化器将激励转换为avro对象，json对象或者字符串，然后结果存储到kafka。 对于接收器连接器，则会发生相反的过程，当worker从kafka读取一条记录时，它使用的配置的转化器将记录从kafka的格式中转换。即连接数据API记录，然后将其传递给接收器，接收器将其插入目标系统。 这允许connect API支持不同类型的数据存储在kafka中，独立于连接器的实现，任何连接器都可以用于任何记录类型，只要有转换器可用。

Offset management offset管理

offset管理式worker为连接器执行的一项便捷的服务API，除了通过rest执行的部署和配置管理之外。其思想是连接器需要知道子集已经处理了哪些数据，并且可以使用kafka提供的api来维护已经处理了的哪些事件信息。 对于源来你借钱，这意味着连接器返回给connect worker的激励包括一个逻辑分区和一个逻辑offset。这些不是kafka分区和kafka的offset。而是源系统中需要的分区和offset。例如，在文件源中，分区可以是文件，offset泽斯文件中的行号或者字符号。在jdbc源中，分区可以是数据库表，而offset可以是表中的激励的id。在编写源连接器时涉及的最重要的涉及决策之一是缺点一种对源系统中的数据进行分区和跟踪offset的好办法。这将影响连接器能够实现的并行级别，以及它是能够提供最少一次还是精确一次的语义。 当源连接器返回记录列表时，其中包括每条记录的源分区和offset。工作人员将这些记录发送给kafka的broker。如果broker成功地确认了这些记录。worker就会存储它并发送kafka的offset。存储机制是支持可插拔的kafka topic。这允许连接器在重启或者崩溃之后从最近存储的offset开始处理消息。 接收连接器有一个相反但是相似的工作流。他们读取kafka的记录，这些记录已经有了一个topic，分区和offset，然后调用连接器的put方法，该方法应该将这些记录存储在目标系统中，如果连接器报告成功，他们就会使用通常的消费者提交的方法，将给连接器的offset重新提交回kafka。 框架本身提供的offset跟踪应该使开发人员更容易编写连接器，并保证在使用不同连接器时在某种程度上保持一致的行为。

Alternatives to Kafka Connect kafka连接器的备选方案

到目前为止，虽然我们已经非常详细地介绍了kafka的connect api，虽然我们喜欢connect api提供的便利和可靠性。但是他们不是唯一的输入和输出kafka数据的方法。让我们看看其他的选择以及他们通常在什么时候被使用.

Ingest Frameworks for Other Datastores 其他数据钻取的框架

我们虽然认为kafka是中心，但是有些人并不同意。有些人围绕hadoop或Elasticsearch之类的系统构建大数据架构。这些系统有自己的数据采集工具，flume之于hadoop和logstash、fluentd之于Elasticsearch。当kafka时架构的不可分割的一部分的时候，并且目标时连接大量的源和输出时，我们推荐kafka的connect API，如果你针对在构建hadoop为中心或者elastic search 为中心的系统，kafka只是众多该系统的组件之一，那么使用flume或者logs’ta’sh更有意义。

GUI-Based ETL Tools 基于GUI的etl工具

从Informatica这样的老式系统，到Talend这也的开源替代产品，Pentaho以及Apache Nifi 和streamset等更新的替代产品都提供了对于apache kafka做为数据源和目的地的支持。如果你以及在使用这些系统，那么使用他们式有意义的，丽日，如果你已经使用了pentaho完成了所有的工作，那么你可能不会对添加另外要给数据集成系统kafka感兴趣。如果你使用的基于gui的方法来构建etl管道，那么它们就很有意义。这些系统的主要缺点式，它们通常式为涉及的工作流而构建的，如果你想要做的只是从kafka种获取数据，那么它们将式一个有些沉重和复杂的解决方案，正如在前文种转换一节所提到的，我们认为数据集成应该专注在所有条件下忠实的传递消息，而大多数etl工具增加了不必要的复杂性。 我们鼓励你将kafka看作一个可同时处理两种数据集成的平台，（通过连接器），应用程序集成（生产者/消费者）和流式处理。kafka可能式ETL工具的一个可行的替代，它只是集成了数据存储。

Stream-Processing Frameworks 流式处理框架

大多数流式处理框架都包含读取实时消息的能力，并将它们写入到其他的一些系统中。如果你的目标系统得到了支持，并且你已经打算使用流处理框架来处理来自kafka的消息，那么使用相同的框架进行数据集成也是合理的。这通常在流式处理工作流中节省了一个步骤（不在kafka中存储处理过的消息，只需要将它们读出兵将其写入到另外一个系统），其缺点式更难的排除诸如丢失和损坏消息之类的问题。

Summary 总结

在本章中，我们讨论了kafka在数据集成中的使用，从使用kafka进行数据集成的原因开始，我们讨论了数据集成解决方案的一般考虑事项。我们展示了为什么我们认为kafka和它的connect api式一个很好的选择，然后我们给出了几个如何在不同场景中使用kafka connect的例子，花了一些时间差康connect是如何工作的，然后讨论了kafka connecct的一些替代方案。 无论你采用何种数据集成的解决方案，最重要的特性始终是它能够在所有故障条件下交付所有消息。我们相信kafka connet是非常可靠的。基于它与kafka久经考验的可靠性的整合。但是重要的是你要测试你所选择的系统，向我们做的一样，确保你选择的数据集成系统能够在进程停止、机器宕机、网络延迟和高负载的情况下存活而不会丢失任何消息，毕竟，数据集成系统只有一个任务就是交付所有的消息。 当然，在集成数据系统的时候，可靠性是最重要的需求，但是它只是要给需求。在选择数据系统的时候，首先你需要检查你的需求是很重要的。参见构建时的考虑事项。然后确保你选择的系统能满足他们。但是这还不够，你还必须充分连接你的数据集成解决方案，以确保你使用它的方式能够支持你的需求。kafka支持至少一次语义是不够的，你必须确保他不会被意外的某种方式配置它，从而导致可靠性过低。