Set up docs MCPApache 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 からアクセスできるフォルダーに配置し、絶対パスを控えておきます
[object Object]
-
Apache NiFi で Controller Service を設定するには、"gear" ボタンをクリックして NiFi Flow Configuration ページを開きます
-
Controller Services タブを選択し、右上の
+ボタンをクリックして新しい Controller Service を追加します
-
DBCPConnectionPoolを検索し、"Add" ボタンをクリックします
-
新しく追加した
DBCPConnectionPoolは、デフォルトでは Invalid 状態です。設定を開始するには、"gear" ボタンをクリックします
-
"Properties" セクションで、以下の値を入力します
| プロパティ | 値 | 備考 |
|---|---|---|
| Database Connection URL | jdbchttps://HOSTNAME:8443/default?ssl=truetrue | 接続 URL 内の HOSTNAME を適宜置き換えてください |
| Database Driver Class Name | com.clickhouse.jdbc.ClickHouseDriver | |
| Database Driver Locations | /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 セクションで、識別しやすいように Controller Service の名前を "ClickHouse JDBC" に変更します
-
"lightning" ボタンをクリックし、続けて "Enable" ボタンをクリックして
DBCPConnectionPoolController Service を有効にします
-
Controller Services タブを確認し、Controller Service が有効になっていることを確認します
[object Object]
-
適切な上流側および下流側のプロセッサとともに、
ExecuteSQLプロセッサを追加します
-
ExecuteSQLプロセッサの "Properties" セクションで、次の値を入力しますプロパティ 値 備考 Database Connection Pooling Service ClickHouse JDBC ClickHouse 用に設定した Controller Service を選択します SQL select query SELECT * FROM system.metrics ここにクエリを入力します -
ExecuteSQLプロセッサを開始します
-
クエリが正常に処理されたことを確認するには、出力キュー内の
FlowFileのいずれか 1 つを確認します
-
出力
FlowFileの結果を表示するには、表示を "formatted" に切り替えます
[object Object]
-
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プロセッサの input と output を確認します。output は複数の input レコードからなる配列である点に注意してくださいInput
Output
-
PutDatabaseRecordプロセッサの "Properties" セクションで、以下の値を入力しますProperty Value Remark Record Reader JSONTreeReader適切な record reader を選択します Database Type Generic デフォルトのままにします Statement Type INSERT Database Connection Pooling Service ClickHouse JDBC ClickHouse の controller service を選択します Table Name tbl ここにテーブル名を入力します Translate Field Names false 挿入されるフィールド名がカラム名と一致するよう、"false" に設定します Maximum Batch Size 1000 insert あたりの最大行数です。この値は、 MergeRecordプロセッサの "Minimum Number of Records" より小さくしないでください -
各 insert に複数の行が含まれていることを確認するには、テーブルの行数が
MergeRecordで定義した "Minimum Number of Records" の値以上ずつ増えていることを確認します。
-
これで、Apache NiFi を使用して ClickHouse にデータを正常にロードできました！