将文件中的数据推送topic:connect-test中

配置connect-file-source.properties

cat connect-file-source.properties
name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test.txt
topic=connect-test

topic的偏移量存储在/tmp/connect.offsets这个文件中,在config/connect-standalone.properties配置,每次connect启动的时候会根据connector的name获得topic偏移量,然后在继续读取或者写入数据

查看文件test.txt中的数据

cat test.txt

hello 
kafka
hadoop

在使用file source connector时,connector会监听配置的数据文件,如果文件发生变化,例如:追加内容,connector会及时处理新增的数据

启动kafka connect

bin/connect-standalone.sh config/connect-standalone.properties \
 config/connect-file-source.properties

查看推送到topic connect-test 的数据

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

{"schema":{"type":"string","optional":false},"payload":"hello "}
{"schema":{"type":"string","optional":false},"payload":"kafka"}
{"schema":{"type":"string","optional":false},"payload":"hadoop"}

修改connect-file-source.properties,对数据进行转换,配置如下

name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test.txt
topic=connect-test

transforms=MakeMap, InsertSource
transforms.MakeMap.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.MakeMap.field=line
transforms.InsertSource.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertSource.static.field=data_source
transforms.InsertSource.static.value=test-file-source

给test.txt增加数据

cat test.txt 
hello 
kafka
hadoop
jafja
spark
hive

启动kafka connect

bin/connect-standalone.sh config/connect-standalone.properties \
 config/connect-file-source.properties

查看推送到topic connect-test 的数据

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

{"schema":{"type":"string","optional":false},"payload":"hello "}
{"schema":{"type":"string","optional":false},"payload":"kafka"}
{"schema":{"type":"string","optional":false},"payload":"hadoop"}
{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"line"},{"type":"string","optional":true,"field":"data_source"}],"optional":false},"payload":{"line":"jafja","data_source":"test-file-source"}}
{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"line"},{"type":"string","optional":true,"field":"data_source"}],"optional":false},"payload":{"line":"spark","data_source":"test-file-source"}}
{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"line"},{"type":"string","optional":true,"field":"data_source"}],"optional":false},"payload":{"line":"hive","data_source":"test-file-source"}}

将topic connect-test 的数据保存到文件中

配置cat config/connect-file-sink.properties

name=local-file-sink
connector.class=FileStreamSink
tasks.max=1
file=test.sink.txt

启动kafka connect

bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-sink.properties

查看生成的test.sink.txt

cat test.sink.txt 
hello 
kafka
hadoop
Struct{line=jafja,data_source=test-file-source}
Struct{line=spark,data_source=test-file-source}
Struct{line=hive,data_source=test-file-source}

启动多个connector

bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties