概要

Confluent for Kubernetes （CFK）は、プライベートクラウド環境（今回は Azure Kubernetes Service（AKS））に Confluent をデプロイして管理するためのクラウドネイティブのコントロールプレーンです。宣言型 API で Confluent Platform をカスタマイズ、デプロイ、管理するための標準的で簡素なインターフェイスが備わっています。

CFK を使用して AKS上にデプロイされている Confluent Platform にこの記事を参考にして下図のような構成を組みます

ソース側としてのCosmosDBに対してデータを書き込み、Confluent Platform 上の topic を経由して、シンク側のCosmosDBにそのままのデータが蓄積されることを確認します。KSQLによるデータ抽出等を実装する前の単純な確認作業となります。

ローカル環境

macOS Monterey 12.3.1
python 3.8.12
Azure CLI 2.34.1
helm v3.6.3
kubectl v1.21.3

事前準備

この記事を参考にして、上図の構成が組まれていること

Sink CosmosDB 設定項目	値
Endpoint	https://iturucosmosdb01.documents.azure.com:443/
CosmosDB アカウント名	iturucosmosdb01
データベース名	CPDemoDB002
コンテナ名	container002
パーティション	section

Source CosmosDB 設定項目	値
Endpoint	https://iturucosmosdb01.documents.azure.com:443/
CosmosDB アカウント名	iturucosmosdb01
データベース名	CPDemoDB001
コンテナ名	container001
パーティション	section

Simpleテストの実行

ソース側 CosmosDB へのデータ書き込み

この記事にあるデータ生成プログラムを利用して、ローカル環境からソース側 CosmosDB にデータを書き込みます（連続で８件のデータを生成）。

$ python cosmosdb_IoTdummy.py --mode db --db write --wait 0 --count 8
データベースへのデータ書き込み
['1657199670.821585', 0, '111', 'O', '432-2664', '福井県', 107.22296451311148, 69.81618009103482, '2022-07-07T22:14:30.821725']
['1657199670.821737', 1, '111', 'C', '118-3635', '沖縄県', 121.79679326570849, 77.41292379149846, '2022-07-07T22:14:30.821773']
['1657199670.82178', 2, '111', 'M', '342-8730', '埼玉県', 148.90352784331895, 59.66836612610244, '2022-07-07T22:14:30.821807']
['1657199670.821812', 3, '111', 'Z', '065-7284', '香川県', 160.81864815360737, 87.7771508916959, '2022-07-07T22:14:30.821838']
['1657199670.821843', 4, '111', 'X', '267-3435', '新潟県', 140.15235764972246, 71.62379403858466, '2022-07-07T22:14:30.821868']
['1657199670.821873', 5, '111', 'D', '406-8826', '島根県', 144.91235299507284, 51.99074105853973, '2022-07-07T22:14:30.821897']
['1657199670.821901', 6, '111', 'A', '133-0863', '佐賀県', 150.37960537206877, 53.52995764628445, '2022-07-07T22:14:30.821926']
['1657199670.82193', 7, '111', 'Z', '121-8609', '大分県', 136.44769557485782, 58.31648238948629, '2022-07-07T22:14:30.821954']

Inserted 8 row(s) of data.
Done.

処理時間:1.1323838233947754 [sec]

Simpleテストの結果

Confluent Control Center 上での確認

Confluent Control Center にログインします。

$ kubectl confluent dashboard controlcenter
http://localhost:9021

該当トピック「topic002」のメッセージを確認します。結果は想定の通りとなっています。

合わせて、Schema情報（value / key）を確認します。結果は想定の通りとなっています。

Azure Portal 上での確認

Azure Portalにサインインし、該当のCosmosDBリソースにアクセスします。

ソース側のCosmosDB「CPDemo001/container001」の「items」を確認します。結果は想定の通りとなっています。

シンク側のCosmosDB「CPDemo002/container002」の「items」を確認します。結果は想定の通り上記と同じデータが同じ順番で蓄積されています。

まとめ

Confluent Platform 上に構成された「Source Connector」→「topic」→「Sink Connector」の経路を通って、データが問題なく流れていることを確認できました。これで、KSQLの実装に進むことができます。