kafka(3) -- 连接导入、导出数据

有些场景下Kafka需要使用其他来源的数据或导出Kafka的数据到其他系统,相对于许多系统需要编写定制集成的代码,使用Kafka连接到系统去导入或导出数据更加简单。
Kafka Connect是包括在Kafka中一个工具,用来导入导出数据到Kafka。它是connectors的一个可扩展工具,其执行定制逻辑,用于与外部系统交互。本文介绍如何使用Kafka Connect做一些简单的连接器从一个文件导入数据到Kafka的主题,和将主题数据导出到另一个文件,具体操作如下:
1. 创建原始数据文件

echo -e "foo/bar" > test.txt#文件名不要修改

test.txt文件中的内容即要导入到kafka中的数据。
2. 启动连接器
启动两个运行在独立模式的连接器,连接器在一个单一的,局部的,专用的进程中运行。需要提供三个配置文件作为参数。第一个参数为Kafka连接过程中的公共配置文件,如要连接到的Kafka的代理服务器的配置和数据的序列化格式的配置。其余两个配置文件用来创建指定的连接器。这些文件包括一个唯一的连接器名称,需要实例化的连接器类,还有创建该连接器所需的其他配置:
bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

用这些Kafka的示例配置文件,使用前面已经启动的本地群集的默认配置,建立两个连接器:第一是一个源连接器,其从输入文件中读取每行的内容,发布到的Kafka主题和第二个是一个sink连接器负责从Kafka主题读取消息,生产出的消息按行输出到文件。
【kafka(3) -- 连接导入、导出数据】3. 检查数据导入、导出
在启动过程中可以看到连接器被实例化的信息,一旦Kafka Connect进程已经起来,源连接器会从test.txt读取每行的消息,并将其生产发布到主题connect-test,而sink连接器会从主题connect-test读取消息,并将其写入文件test.sink.txt。可以通过检查输出文件的内容来验证数据都已通过整个管道输送:
#cat test.sink.txt foo bar

由于数据被存储在Kafka topic connect-test中,所以我们也可以运行控制台消费者消费topic中的数据或者自定义消费者逻辑消费消息:
# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning {"schema":{"type":"string","optional":false},"payload":"foo"} {"schema":{"type":"string","optional":false},"payload":"bar"} ...

由于连接器不停的处理数据,因此我们可以继续将数据添加到test.txt文件,并能看到数据通过管道移动:
echo 111 >> test.txt cat test.sink.txt

下一篇: 使用Streams处理数据

    推荐阅读