网站建优化重庆百度小额贷款有限公司

一、上下文

《Kafka-Connect》中已经阐述了Kafka-Connect的理论知识，为了更生动的理解它，我们今天通过官方的一个小例子来感受下它的妙用。

二、创建topic

kafka-topics --create --topic connect-test --bootstrap-server cdh1:9092 --partitions 2 --replication-factor 2

三、编写配置文件

在cdh环境中，这些配置文件所在的目录为：

/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/

1、connect-standalone.properties

# 用于建立与Kafka集群的初始连接的主机/端口对列表。以下是cdh中的例子
bootstrap.servers=cdh1:9092,cdh2:9092,cdh3:9092

# 转换器指定Kafka中数据的格式以及如何将其转换为Connect数据。每个Connect用户都需要根据他们希望从Kafka加载或存储数据时使用的格式进行配置
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# 转换器特定的设置可以通过在转换器的设置前加上我们想要应用的转换器来传递
key.converter.schemas.enable=true
value.converter.schemas.enable=true

offset.storage.file.filename=/tmp/connect.offsets
# 刷新速度比正常情况快得多，这对测试/调试很有用
offset.flush.interval.ms=10000

# 设置为用逗号（，）分隔的文件系统路径列表，以启用插件（连接器、转换器、转换）的类加载隔离。该列表应由顶级目录组成，其中包括以下内容的任意组合：
# a) 直接包含带有插件及其依赖项的jar的目录
# b) uber包含插件及其依赖项
# c) 直接包含插件类及其依赖项的包目录结构的目录
# 注意：将遵循符号链接来发现依赖关系或插件。
# Examples: 
# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
#plugin.path=

2、connect-file-source.properties

name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/connect-file-test-data/source.txt
topic=connect-test

3、connect-file-sink.properties

name=local-file-sink
connector.class=FileStreamSink
tasks.max=1
file=/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/connect-file-test-data/sink.txt
topics=connect-test

四、运行

cd /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/
./connect-standalone.sh /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/connect-standalone.properties /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/connect-file-source.properties /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/connect-file-sink.properties

 启动成功后如图所示：

五、测试

我们项source.txt 中写入一些数据

echo 1 >> source.txt
echo 2 >> source.txt
echo 3 >> source.txt
echo 4 >> source.txt
echo 5 >> source.txt
echo 6 >> source.txt
echo 7 >> source.txt
echo 8 >> source.txt
echo 9 >> source.txt
echo 10 >> source.txt

从结果上看感觉sink.txt中结果是乱序的，这是因为我们创建topic时设置了2个分区，我们用consoumer来看看各个分区的顺序情况：

kafka-console-consumer --topic connect-test --from-beginning --bootstrap-server cdh1:9092,cdh2:9092,cdh3:9092 --partition 0

kafka-console-consumer --topic connect-test --from-beginning --bootstrap-server cdh1:9092,cdh2:9092,cdh3:9092 --partition 1

 

我们再次看sink.txt的结果就可以理解了，kafka只保证了分区有序，如果使用Kafka-Connect时想保证文件的输入和输出是有序的，就需要设定topic为1个分区。