Apache NiFiをClickHouseに接続する

Community Maintained

Apache NiFi は、ソフトウェアシステム間のデータフローを自動化するために設計されたオープンソースのワークフロー管理ソフトウェアです。ETLデータパイプラインの作成が可能で、300以上のデータプロセッサが同梱されています。このステップバイステップのチュートリアルでは、Apache NiFiをソースと宛先の両方としてClickHouseに接続し、サンプルデータセットをロードする方法を説明します。

接続情報を収集する

HTTP(S) で ClickHouse に接続するには、次の情報が必要です。

Parameter(s)	Description
`HOST` and `PORT`	通常、TLS を使用する場合のポートは 8443、TLS を使用しない場合のポートは 8123 です。
`DATABASE NAME`	既定で `default` という名前のデータベースが用意されています。接続したいデータベースの名前を使用してください。
`USERNAME` and `PASSWORD`	既定のユーザー名は `default` です。用途に応じて適切なユーザー名を使用してください。

ClickHouse Cloud サービスに関する詳細情報は、ClickHouse Cloud コンソールで確認できます。サービスを選択し、Connect をクリックします。

HTTPS を選択します。接続情報は、サンプルの curl コマンド内に表示されます。

自己管理型の ClickHouse を使用している場合、接続情報は ClickHouse 管理者によって設定されます。

Apache NiFi をダウンロードして実行する

新規セットアップの場合は、https://nifi.apache.org/download.html からバイナリをダウンロードし、./bin/nifi.sh start を実行して起動します

ClickHouse JDBC ドライバをダウンロードする

GitHub 上の ClickHouse JDBC ドライバのリリースページにアクセスし、最新の JDBC リリースバージョンを探します
リリースバージョンで「Show all xx assets」をクリックし、「shaded」または「all」というキーワードを含む JAR ファイル（例: clickhouse-jdbc-0.5.0-all.jar）を探します。
JAR ファイルを Apache NiFi からアクセス可能なフォルダに配置し、その絶対パスを記録しておきます

`DBCPConnectionPool` コントローラサービスを追加し、プロパティを設定する

Apache NiFi でコントローラサービスを設定するには、歯車アイコン（"gear" ボタン）をクリックして NiFi Flow Configuration ページを開きます
Controller Services タブを選択し、右上の + ボタンをクリックして新しいコントローラサービスを追加します
DBCPConnectionPool を検索し、「Add」ボタンをクリックします
追加したばかりの DBCPConnectionPool は、デフォルトでは無効 (Invalid) 状態になっています。歯車アイコン（"gear" ボタン）をクリックして設定を開始します
「Properties」セクションで、次の値を入力します

Property	Value	Remark
Database Connection URL	jdbc:ch:https://HOSTNAME:8443/default?ssl=true	接続 URL 中の HOSTNAME を環境に合わせて置き換えます
Database Driver Class Name	com.clickhouse.jdbc.ClickHouseDriver
Database Driver Location(s)	/etc/nifi/nifi-X.XX.X/lib/clickhouse-jdbc-0.X.X-patchXX-shaded.jar	ClickHouse JDBC ドライバ JAR ファイルへの絶対パス
Database User	default	ClickHouse ユーザー名
Password	password	ClickHouse パスワード

Settings セクションで、コントローラサービスの名前を分かりやすくするために「ClickHouse JDBC」に変更します
「lightning」ボタンをクリックし、続いて「Enable」ボタンをクリックして DBCPConnectionPool コントローラサービスを有効化します
Controller Services タブを開き、コントローラサービスが有効化されていることを確認します

`ExecuteSQL` プロセッサを使用してテーブルから読み取る

適切なアップストリームおよびダウンストリームのプロセッサと共に ExecuteSQL プロセッサを追加します

ExecuteSQL プロセッサの「Properties」セクションで次の値を入力します

Property	Value	Remark
Database Connection Pooling Service	ClickHouse JDBC	ClickHouse 用に設定した Controller Service を選択します
SQL select query	SELECT * FROM system.metrics	ここにクエリを入力します

ExecuteSQL プロセッサを起動する
クエリが正常に処理されたことを確認するには、出力キュー内の FlowFile のいずれかを確認してください
出力された FlowFile の結果を確認するには、ビューを「formatted」表示に切り替えます

`MergeRecord` および `PutDatabaseRecord` プロセッサを使用してテーブルに書き込む

1 回の insert で複数の行を書き込むには、まず複数のレコードを 1 つのレコードに結合する必要があります。これは MergeRecord プロセッサを使用することで実現できます。

MergeRecord プロセッサの「Properties」セクションで、次の値を入力します

Property	Value	Remark
Record Reader	`JSONTreeReader`	適切な Record Reader を選択します
Record Writer	`JSONReadSetWriter`	適切な Record Writer を選択します
Minimum Number of Records	1000	1 つのレコードを構成するためにマージされる行の最小数を増やすよう、この値をより大きな値に設定します。デフォルトは 1 行です
Maximum Number of Records	10000	"Minimum Number of Records" より大きな値に設定します。デフォルトは 1,000 行です

複数のレコードが 1 つにマージされていることを確認するには、MergeRecord プロセッサの入力と出力を確認してください。出力は複数の入力レコードから成る配列となる点に注意してください

入力

出力

PutDatabaseRecord プロセッサの「Properties」セクションで、次の値を入力します

Property	Value	Remark
Record Reader	`JSONTreeReader`	適切なレコードリーダーを選択します
Database Type	Generic	デフォルトのままにします
Statement Type	INSERT
Database Connection Pooling Service	ClickHouse JDBC	ClickHouse コントローラサービスを選択します
Table Name	tbl	ここにテーブル名を入力します
Translate Field Names	false	挿入されるフィールド名がカラム名と一致するように、"false" に設定します
Maximum Batch Size	1000	1 回の INSERT あたりの最大行数。この値は、`MergeRecord` プロセッサの "Minimum Number of Records" の値より小さくしないでください

各挿入に複数の行が含まれていることを確認するには、テーブルの行数が MergeRecord で定義されている「Minimum Number of Records」の値以上ずつ増加していることを確認します。
おめでとうございます。Apache NiFi を使用して ClickHouse へのデータ取り込みに成功しました！

接続情報を収集する​

Apache NiFi をダウンロードして実行する​

ClickHouse JDBC ドライバをダウンロードする​

DBCPConnectionPool コントローラサービスを追加し、プロパティを設定する​

ExecuteSQL プロセッサを使用してテーブルから読み取る​

MergeRecord および PutDatabaseRecord プロセッサを使用してテーブルに書き込む​