Apache NiFi と Hologres の接続方法 - Hologres - Alibaba Cloud ドキュメントセンター

このドキュメントでは、Apache NiFi を Hologres に接続する方法の例を示します。

背景情報

Apache NiFi は、データ処理とディストリビューションのための、使いやすく信頼性の高いシステムです。システム間のデータフローを自動化するように設計されています。Apache NiFi は、システム間またはシステム内のデータストリームを管理および処理するための、高度にインタラクティブでユーザーフレンドリな Web インターフェイスを提供します。

前提条件

Hologres インスタンスが有効化されていること。詳細については、「Hologres インスタンスの購入」をご参照ください。
Apache NiFi がインストールされていること。詳細については、「NiFi のインストールと起動方法」をご参照ください。

ローカル JSON ファイルの Hologres への書き込み

次のワークフローは、ローカル JSON ファイルから Hologres にデータを書き込みます。

GetFile：JSON ファイルを読み取ります。
ConvertJSONToSQL：JSON ファイル内の要素を SQL INSERT 文に変換します。
PutSQL：前のプロセッサからの SQL ステートメントを実行して、データベースにデータを挿入します。

データベースとテーブルの作成
1. ご利用の Hologres インスタンスにログインし、demo という名前のデータベースを作成します。詳細については、「データベースの作成」をご参照ください。
2. テーブルの作成
  次の SQL ステートメントを実行して、送信先テーブルを作成します。
```
DROP TABLE IF EXISTS user_info;
CREATE TABLE IF NOT EXISTS user_info (
    id int,
    first_name text,
    last_name text,
    email text
);
```
GetFile プロセッサの構成
1. GetFile プロセッサの追加
  
  詳細については、「プロセッサの追加」をご参照ください。
2. JSON ファイルへのパスの指定
  [プロパティ] タブで、[入力ディレクトリ] フィールドにご利用の JSON ファイルのストレージパスを入力します。この例では、Apache NiFi サーバーの /opt/nifi/nifi-current/file_source ディレクトリに保存されている user_info.json という名前の JSON ファイルを使用します。ファイルには次のデータが含まれています：
```
{
    "id": 1,
    "first_name": "Sig",
    "last_name": "Olivo",
    "email": "solivo0@blinklist.com"
}
```
  以下は設定例です。他のプロパティはデフォルト値のままにします：[ファイルフィルター] は [^\.].*、[ソースファイルを保持] は false (読み取り後にソースファイルを削除)、[バッチサイズ] は 10、[サブディレクトリを再帰的に処理] は true、[ポーリング間隔] は 0 sec です。
3. Apply をクリックして設定を保存します。

ConvertJSONToSQL プロセッサの構成

ConvertJSONToSQL プロセッサを追加します。
[JDBC 接続プール] プロパティで、新しいサービスを作成します。[互換性のあるコントローラーサービス] を DBCPConnectionPool に、[コントローラーサービス名] を hologres に設定します。
[JDBC 接続プール] 行の右端にある移動矢印 (→) をクリックして、接続文字列を設定します。
先ほど作成した DBCPConnectionPool を見つけ、その設定アイコンをクリックします。

[コントローラーサービス] タブで、hologres という名前の DBCPConnectionPool サービスを見つけ、右側にある設定アイコン (歯車) をクリックします。

設定ページの [プロパティ] タブで、次のパラメーターを設定します。

パラメーター	説明	説明
データベース接続 URL	Hologres インスタンスの JDBC 接続文字列。フォーマットは `jdbc:postgresql://<endpoint>/<database name>` です。例：`jdbc:postgresql://hgpostcn-cn-xxxxxxxxxxx-cn-shanghai.hologres.aliyuncs.com:80/demo`。	エンドポイントは、パブリックエンドポイントまたは Virtual Private Cloud (VPC) エンドポイントである必要があります。エンドポイントを取得するには、Hologres コンソールでインスタンスの詳細ページに移動します。
データベースドライバークラス名	org.postgresql.Driver	該当なし
データベースドライバーの場所	PostgreSQL JDBC ドライバーが保存されているパス。例：`/opt/nifi/nifi-current/jdbc_driver/postgresql-42.3.4.jar`。	JDBC ドライバーは PostgreSQL の公式サイトからダウンロードできます。JDBC ドライバーのバージョン 42.2.25 以降の使用を推奨します。
データベースユーザー	ご利用の Alibaba Cloud アカウントの AccessKey ID。	AccessKey ID を取得するには、AccessKey 管理に移動します。
パスワード	ご利用の Alibaba Cloud アカウントの AccessKey Secret。

[OK] をクリックして設定を完了します。
[有効化] をクリックしてコントローラーサービスを開始します。

ConvertJSONToSQL プロセッサの設定に戻り、次のパラメーターを設定します。詳細については、NiFi の公式ドキュメントをご参照ください。

パラメーター	説明
ステートメントタイプ	生成する SQL ステートメントのタイプ。この例では、`INSERT` を使用します。
テーブル名	送信先テーブルの名前。この例では、`user_info` を使用します。
スキーマ名	送信先テーブルのスキーマ。この例では、`public` を使用します。

Apply をクリックして設定を完了します。

PutSQL プロセッサの構成
1. PutSQL プロセッサを追加します。
2. [JDBC 接続プール] を、前のステップで設定した DBCPConnectionPool に設定します。この例では、DBCPConnectionPool の名前は hologres です。
3. [フラグメント化されたトランザクションのサポート] を false に設定します。
4. Apply をクリックして設定を完了します。
データ書き込みの開始

これで設定は完了です。すべてのプロセッサを開始して JSON ファイルを読み取り、そのデータを Hologres に書き込みます。NiFi データフローパイプラインは、順番に接続された 3 つのプロセッサで構成されています：GetFile (ローカル JSON ファイルを読み取り、105 バイトを出力) → ConvertJSONToSQL (JSON を SQL ステートメントに変換し、83 バイトを出力) → PutSQL (SQL ステートメントを実行してデータベースに書き込み)。実行後、`success` と `sql` のキューが空になっていれば、データが正常に Hologres に書き込まれたことを示します。
データのクエリ
Hologres で次のコマンドを実行して user_info テーブルをクエリし、インポートされたデータを表示します。
```
SELECT * FROM user_info;
```
クエリは 4 つの列 (id、first_name、last_name、email) と 1 行のデータ (id=1, first_name=Sig, last_name=Olivo, email=solivo0@blinklist.com) を返します。この結果は、ローカル JSON ファイルのデータが正常に Hologres に書き込まれたことを示しています。