Apache NiFiによるOB Cloudの統合

Apache NiFiは、データストリーム処理を自動化するプラットフォームであり、システム間で効率的かつ信頼性の高いデータ転送を実現します。OB CloudのBinlogデータを読み取り、ファイルシステム、メッセージキュー（Kafkaなど）、またはHTTPエンドポイントなどのターゲットにデータを配信できます。

前提条件

1. 既にOB Cloudクラウドデータベースアカウントを登録し、以下のリソースを作成していること：

   • クラスタインスタンスとMySQL互換モードのテナントを作成します。詳細については、クラスタインスタンスの作成およびテナントの作成をご参照ください。

   • データベースとアカウント（データベースユーザー）を作成します。詳細については、データベースの作成およびアカウントの作成をご参照ください。

   • MySQL互換モードのテナントでBinlogサービスを購入しています。詳細については、Binlogサービスの購入をご参照ください。

2. NiFiサーバーのIPアドレスがOB Cloudクラスタインスタンスのホワイトリストに追加されていること。詳細については、テナントのホワイトリストグループの設定をご参照ください。

3. ご利用の実行環境がJava Development Kit (JDK) 1.8以上を満たしていること。

4. Apache NiFiインストールパッケージをダウンロード済みであること。

ステップ1：NiFiの実行環境を構成する

1. NiFiインストールパッケージをターゲットディレクトリに解凍します。

2. 以下の方法でconf/nifi.properties設定ファイルを変更します（本記事ではHTTP接続を例に説明します）：

   # HTTPSを無効化し、HTTPアクセスを設定する
   nifi.web.https.host=
   nifi.web.https.port=
   nifi.web.http.host=0.0.0.0
   nifi.web.http.port=<カスタムポート番号>
   
   # 非セキュアなリモート接続を許可する
   nifi.remote.input.secure=false
   

3. MySQLドライバーをNiFiサーバーにアップロードします。

4. NiFiのインストールディレクトリで以下のコマンドを実行してサービスを起動します。サービスを停止する場合は、bin/nifi.sh stopを実行します。

   # サービスを起動する
   bin/nifi.sh start
   

ステップ2：データストリーム処理パイプラインの設定

1. ブラウザで http://<nifi_server_ip>:<port>/nifi/ にアクセスして、NiFiコンソールに入ります。

2. CaptureChangeMySQL Processorを設定します。

   1. ページ上部のProcessorアイコンをキャンバスにドラッグし、CaptureChangeMySQL を検索して選択し、ADD をクリックします。

      

   2. CaptureChangeMySQL Processorを右クリックして Configure を選択し、PROPERTIES タブで関連パラメータを設定した後、APPLY をクリックします。

      プロパティ名	説明
      MySQL Nodes	OB Cloudインスタンスのアドレスとポート。
      MySQL Driver Class Name	JDBCドライバーのクラス名。com.mysql.cj.jdbc.Driver です。
      MySQL Driver Location	アップロードしたドライバーJARファイルがあるディレクトリ。
      Username	作成したデータベースユーザーのユーザー名。
      Password	データベースユーザーのパスワード。
      Database/Schema Name Pattern	Binlogを読み取るデータベース（正規表現をサポート）。
      Table Name Pattern	Binlogを読み取るテーブル（正規表現をサポート）。
      Include DDL Events	DDL変更イベントをキャプチャするため、true に設定する必要があります。

      

3. PutFile Processorを設定します。

   1. 再度ページ上部のProcessorアイコンをキャンバスにドラッグし、PutFile を検索して選択し、ADD をクリックします。

   2. PutFile Processorを右クリックして Configure を選択し、PROPERTIES タブで Directory プロパティを入力します。Binlogファイルはこのディレクトリに出力されます。完了後、APPLY をクリックします。

      

4. データストリーム接続を確立します。

   1. キャンバスに戻り、CaptureChangeMySQL の矢印をクリックして PutFile を指し、両者を接続します。

   2. 表示された Create Connection ポップアップウィンドウで、ADD をクリックします。

   3. キャンバスで、PutFile Processorを右クリックして Configure を選択し、RELATIONSHIPS タブで failure と success の動作ポリシーを設定します。どちらも terminate を選択します。完了後、APPLY をクリックします。

      

5. データ処理フローを開始します。

   1. キャンバスに戻り、CaptureChangeMySQL Processorを右クリックして Start を選択します。

   2. PutFile Processorでも同様に右クリックして Start を選択します。

ステップ3：データ同期を検証する

1. OB Cloudデータベースにデータを挿入します。

2. PutFileディレクトリにファイルが生成されているか確認します。

   

3. ファイルの内容を確認すると、実行されたSQLステートメントの関連情報を見ることができます。