kafka connect簡単テスト

ファイルのデータを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オフセット量を取得し、データの読み取りまたは書き込みを継続します.

ファイルを表示する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