Apache NiFi を ClickHouse に接続する

​

接続情報を確認する

​

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

​

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

1. GitHub のClickHouse JDBCドライバーのリリースページにアクセスし、最新の JDBC リリースを確認します
2. 該当するリリースで “Show all xx assets” をクリックし、“shaded” または “all” を含む JAR ファイルを探します (例: clickhouse-jdbc-0.5.0-all.jar)
3. その JAR ファイルを Apache NiFi からアクセスできるフォルダーに配置し、絶対パスを控えておきます

​

DBCPConnectionPool Controller Service を追加し、そのプロパティを設定する

1. Apache NiFi で Controller Service を設定するには、“gear” ボタンをクリックして NiFi Flow Configuration ページを開きます
2. Controller Services タブを選択し、右上の + ボタンをクリックして新しい Controller Service を追加します
3. DBCPConnectionPool を検索し、“Add” ボタンをクリックします
4. 新しく追加した DBCPConnectionPool は、デフォルトでは Invalid 状態です。設定を開始するには、“gear” ボタンをクリックします
5. “Properties” セクションで、以下の値を入力します

6. Settings セクションで、識別しやすいように Controller Service の名前を “ClickHouse JDBC” に変更します
7. “lightning” ボタンをクリックし、続けて “Enable” ボタンをクリックして DBCPConnectionPool Controller Service を有効にします
   
8. Controller Services タブを確認し、Controller Service が有効になっていることを確認します

​

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

1. 適切な上流側および下流側のプロセッサとともに、ExecuteSQL プロセッサを追加します
2. ExecuteSQL プロセッサの “Properties” セクションで、次の値を入力します

   プロパティ	値	備考
   Database Connection Pooling Service	ClickHouse JDBC	ClickHouse 用に設定した Controller Service を選択します
   SQL select query	SELECT * FROM system.metrics	ここにクエリを入力します

3. ExecuteSQL プロセッサを開始します
4. クエリが正常に処理されたことを確認するには、出力キュー内の FlowFile のいずれか 1 つを確認します
5. 出力 FlowFile の結果を表示するには、表示を “formatted” に切り替えます

​

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

1. 1 回の insert で複数の行を書き込むには、まず複数のレコードを 1 つのレコードにマージする必要があります。これは MergeRecord プロセッサを使用して行えます
2. 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 行です

3. 複数のレコードが 1 つにマージされていることを確認するには、MergeRecord プロセッサの input と output を確認します。output は複数の input レコードからなる配列である点に注意してください Input Output
4. 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” より小さくしないでください

5. 各 insert に複数の行が含まれていることを確認するには、テーブルの行数が MergeRecord で定義した “Minimum Number of Records” の値以上ずつ増えていることを確認します。
6. これで、Apache NiFi を使用して ClickHouse にデータを正常にロードできました！