KafkaConnect导入导出数据

一、Kafka Connect 导入/导出 数据

从控制台写入和写回数据是一个方便的开始，但你可能想要从其他来源导入或导出数据到其他系统。对于大多数系统，可以使用kafka Connect，而不需要编写自定义集成代码。

Kafka Connect是导入和导出数据的一个工具。是一个在 kafka与其他系统之间的可扩展和可靠的流数据工具。它使得快速定义连接器变得非常简单，这些连接器可以将大量数据移入和移出Kafka。Kafka Connect
可以摄取整个数据库或者从你所有的应用服务器上收集指标到Kafka主题中，使数据可以用于低延迟的流处理。导出作业可以将数据从Kafka主题交付到二级存储和查询系统，或者交付到批处理系统以进行离线分析。

Kafka Connect功能包括:

• Kafka Connect是Kafka连接器的通用框架——Kafka Connect标准化了其他数据系统与Kafka的集成，简化了连接器的开发、部署和管理
• 分布式和独立模式——向上扩展到支持整个组织的大型集中管理服务，或者向下扩展到开发、测试和小型生产部署
• 通过一个易于使用的REST API向Kafka Connect集群提交和管理连接器
• 自动偏移量管理——Kafka Connect可以自动管理偏移量提交过程，所以连接器开发人员不需要担心连接器开发中这个容易出错的部分
• Kafka Connect建立在现有的组管理协议上。可以添加更多的worker来扩展Kafka Connect集群。
• 流/批处理集成-利用Kafka现有的功能，Kafka Connect是一个理想的解决方案桥接流和批处理数据系统

一些概念

• kafka connector:是kafka connect的关键组成部分，它是一个逻辑上的job,用于在kafka和其他系统之间拷贝数据,比如：从上游系统拷贝数据到kafka,或者从kafka拷贝数据到下游系统

• Tasks:每个kafka connector可以初始化一组task进行数据的拷贝

• Workers:逻辑上包含kafka connector和tasks用来调度执行具体任务的进程,具体执行时分为standalone模式和distributed模式

​

下面介绍配置、运行和管理Kafka Connect。

二、运行 Kafka Connect

Kafka Connect目前支持两种执行模式:单机(单进程)和分布式。

在独立模式下，所有工作都在单个进程中执行。这种配置更容易设置和开始，可能在只有一个worker有意义的情况下有用(例如收集日志文件)，但它没有从Kafka Connect的一些特性中受益，比如容错。可以使用如下命令启动独立进程:

1

bin/connect-standalone.sh config/connect-standalone.properties connector1.properties [connector2.properties ...]

单例配置

配置文件在 config/connect-standalone.properties

首先是Kafka Connect处理的配置，包含常见的配置，例如要连接的Kafka broker和数据的序列化格式。其余的配置文件都指定了要创建的连接器。包括连接器唯一名称，和要实例化的连接器类。以及连接器所需的任何其他配置。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 连接器的唯一名称。再次尝试注册相同名称将会失败。
nname
# kafka 服务器地址
bootstrap.servers=localhost:9092
# 唯一的字符串，用于标识此worker所属的Connect集群组。
# group.id=
# 用于在Kafka连接格式和写入Kafka的序列化格式之间转换的转换器类。
# 控制了从Kafka写入或读取的消息中的键的格式，并且由于它独立于连接器，它允许任何连接器使用任何序列化格式。常见的格式包括JSON和Avro。
key.converter=org.apache.kafka.connect.json.JsonConverter
# 控制了写入或从Kafka读取的消息中的值的格式，并且由于它独立于连接器，它允许任何连接器使用任何序列化格式。常见的格式包括JSON和Avro。
value.converter=org.apache.kafka.connect.json.JsonConverter
# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
# it to
key.converter.schemas.enable=true
value.converter.schemas.enable=true
# 用逗号分隔的主题列表，用作此连接器的输入的话题
# topics
# 主题的Java正则表达式，用作此连接器的输入对于任何其他选项，您应该查阅连接器的文档。
# topics.regex=
# kafka topic仓库配置,需要手动创建主题，以确保正确配置 default connect-configs
# config.storage.topic
offset.storage.file.filename=/tmp/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000

控制台操作

在这个快速入门里，我们将看到如何运行Kafka Connect用简单的连接器从文件导入数据到Kafka主题，再从Kafka主题导出数据到文件。

首先，我们首先创建一些种子数据用来测试：

1

echo -e "foo\nbar" > test.txt

接下来，我们开始2个连接器运行在独立的模式，这意味着它们运行在一个单一的，本地的，专用的进程。我们提供3个配置文件作为参数。

1
2
3

bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties
# 这是分布式的命令，暂时不考虑
bin/connect-distributed.sh config/connect-distributed.properties

这些包含在Kafka中的示例配置文件使用之前启动的默认本地群集配置，并创建两个连接器： 第一个是源连接器，用于从输入文件读取行，并将其输入到 Kafka topic。 第二个是接收器连接器，它从Kafka
topic中读取消息，并在输出文件中生成一行。

在启动过程中，你会看到一些日志消息，包括一些连接器正在实例化的指示。 一旦Kafka Connect进程启动，源连接器就开始从test.txt读取行并且 将它们生产到主题connect-test
中，同时接收器连接器也开始从主题connect-test中读取消息， 并将它们写入文件test.sink.txt中。我们可以通过检查输出文件的内容来验证数据是否已通过整个pipeline进行交付：

1

more test.sink.txt

出现结果

1
2

foo
bar

请注意，数据存储在Kafka topicconnect-test中，因此我们也可以运行一个console consumer（控制台消费者）来查看 topic 中的数据（或使用custom
consumer（自定义消费者）代码进行处理）：

1

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning

结果：

1
2

{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}

连接器一直在处理数据，所以我们可以将数据添加到文件中，并看到它在pipeline 中移动：

1
2

# 追加一行数据
echo Another line >> test.txt

应该可以看到这一行出现在控制台用户输出和接收器文件中。

三、MySql 操作

https://www.pianshen.com/article/3877182847/

四、详细配置

本文地址： https://github.com/maxzhao-it/blog/post/34489/