■ はじめに

https://dk521123.hatenablog.com/entry/2023/04/23/235534 https://dk521123.hatenablog.com/entry/2023/04/26/103421 https://dk521123.hatenablog.com/entry/2023/04/27/235703 https://dk521123.hatenablog.com/entry/2023/04/28/014737

の続き。

今回は、Kafka Connect について、扱う。

【１】Kafka Connect
【２】Kafka Connect の種類
　１）Source
　２）Sink
【３】Kafka Connect の実行モード
　１）スタンドアローンモード
　２）分散モード
【４】Quick Start
　０）実行前の下準備
　１）Source
　補足：設定ファイル
【５】REST API
　１）Source
　２）Sink
　３）Connectorの削除

【１】Kafka Connect

* Kafkaと周辺のシステム間で
　ストリームデータをやりとりするためのツール
 => 英語だけど、Kafka Connect の説明動画あり。

https://developer.confluent.io/learn-kafka/kafka-connect/how-connectors-work/

【２】Kafka Connect の種類

構成図

Producer => [Source] => Kafka => [Sink] => Consumer

１）Source

* Producer （Any to Kafka） 側の Connector

２）Sink

* Consumer （Kafka to any）側の Connector

【３】Kafka Connect の実行モード

１）スタンドアローンモード
２）分散モード

１）スタンドアローンモード

* 全ての仕事が1つのプロセスの中で実施

起動例

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

# [解説]
# * connect-standalone.properties ... Workの設定
# * connector1.properties ... Connectorの設定

# Windows
bin\windows\connect-standalone.bat \
 config\connect-standalone.properties config\connect-file-source.properties

２）分散モード

* 複数台のサーバーで分散処理を実施
* 同じ group.id を使用して複数のワーカープロセスを開始

起動例

# Linux
bin/connect-distributed.sh config/connect-distributed.properties

# Windows
bin\windows\connect-distributed.bat config\connect-distributed.properties

【４】Quick Start

０）実行前の下準備

[1] zookeeper起動

# Start the ZooKeeper service
.\bin\windows\zookeeper-server-start.bat config\zookeeper.properties

[2] Kafka broker起動

# Start the Kafka broker service
.\bin\windows\kafka-server-start.bat config\server.properties

[3] トピックの確認

#  Topic「connect-test」確認
bin\windows\kafka-topics.bat --describe --topic connect-test --bootstrap-server localhost:9092

# もし、何も作成していない場合は、Topic「connect-test」作成
# bin\windows\kafka-topics.bat --create --topic connect-test --bootstrap-server localhost:9092

１）Source

[1] 送信するコンテンツ作成

* Kafkaがインストールしている配下に「test.txt」を置き
　 送信内容を記述し、保存する
~~~~~~~~~~~~
Hello, world!!!?
~~~~~~~~~~~~

[2] SourceとしてのKafka Connectを起動

bin\windows\connect-standalone.bat config\connect-standalone.properties config\connect-file-source.properties

[3] 確認

# test.txt の内容が表示されるか確認
bin\windows\kafka-console-consumer.bat --topic connect-test --from-beginning --bootstrap-server localhost:9092

補足：設定ファイル

config\connect-standalone.properties

# These are defaults. This file just demonstrates how to override some settings.
bootstrap.servers=localhost:9092

# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
# need to configure these based on the format they want their data in when loaded from or stored into Kafka
key.converter=org.apache.kafka.connect.json.JsonConverter
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
offset.storage.file.filename=/tmp/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000

config\connect-file-source.properties

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

config\connect-file-sink.properties

name=local-console-sink
connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector
tasks.max=1
topics=connect-test

【５】REST API

* curlコマンドを使って、REST APIを呼び出し、Source/Sinkをしてみる
 => いつの間にか、Windowsでも使える
 => 「curl --version」で使えるか確認

curl コマンド
https://dk521123.hatenablog.com/entry/2017/12/05/233100

１）Source

[1] Sourceのデータ「sample-source.txt」を作成する

Hello
World!!
Bye

[2] REST APIに送るSource用データ「sample-source.json」を作成する(「connector.class」で「FileStreamSourceConnector」を指定)

{
  "name" : "sample-source-data",
  "config" : {
    "connector.class" : "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "file" : "sample-source.txt",
    "tasks.max" : "1",
    "topic" : "connect-test"
  }
}

[3] REST APIに送る

curl -X POST --header "Content-Type: application/json" http://localhost:8083/connectors -d @sample-source.json

# 「curl http://localhost:8083/connectors/【ConnectionName】/status」で確認可能
curl http://localhost:8083/connectors/sample-source-data/status

補足：コネクターのステータスについて
https://docs.confluent.io/ja-jp/platform/7.1/connect/monitoring.html#connector-and-task-status

Status	Explanations
"state":"UNASSIGNED"	コネクター/タスクは、まだワーカーに割り当てられていない
"state":"RUNNING"	コネクター/タスクが実行中
"state":"PAUSED"	コネクター/タスクが一時停止
"state":"FAILED"	コネクター/タスクが失敗

２）Sink

[1] REST APIに送るSink用データ「sample-sink.json」を作成する(「connector.class」で「FileStreamSinkConnector」を指定)

{
  "name" : "sample-sink-data",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector", 
    "file" : "sample-sink.txt",
    "tasks.max" : "1",
    "topics": "connect-test"
  }
}

[2] REST APIに送る

curl -X POST --header "Content-Type: application/json" http://localhost:8083/connectors -d @sample-sink.json

# 「curl http://localhost:8083/connectors/【ConnectionName】/status」で確認可能
curl http://localhost:8083/connectors/sample-sink-data/status

[3] 動作確認（「sample-sink.txt」の内容が「sample-source.txt」と同じことを確認）

Hello
World!!
Bye

３）Connectorの削除

[1] Connectorの確認

curl http://localhost:8083/connectors

<出力結果>
["sample-source-data","sample-sink-data"]

[2] Connectorの削除

curl -X DELETE http://localhost:8083/connectors/sample-source-data
curl -X DELETE http://localhost:8083/connectors/sample-sink-data

[3] Connectorの再確認

curl http://localhost:8083/connectors

<出力結果>
[]

参考文献

http://mogile.web.fc2.com/kafka/documentation_8.html
https://tutuz-tech.hatenablog.com/entry/2019/03/19/093259
https://tutuz-tech.hatenablog.com/entry/2019/03/21/000835
https://qiita.com/knoguchi/items/d0b8ed1297829dc58346
https://tjtjtj.hatenablog.com/entry/2019/11/20/230000

プログラムの超個人的なメモ

Memo for Programming.

【Kafka】Kafka Connect ～基礎知識編～

■ はじめに

目次

【１】Kafka Connect

【２】Kafka Connect の種類

１）Source

２）Sink

【３】Kafka Connect の実行モード

１）スタンドアローンモード

２）分散モード

【４】Quick Start

０）実行前の下準備

１）Source

補足：設定ファイル

【５】REST API

１）Source

２）Sink

３）Connectorの削除

参考文献

関連記事