Openflow Connector for MySQL を設定する

注釈

コネクタには、 コネクタ利用規約 が適用されます。

このトピックでは、 Openflow Connector for MySQL を設定する手順について説明します。

前提条件

  1. Openflow Connector for MySQL について を確認してください。

  2. Snowflakeとデータを同期するため、 MySQL 8以降のバージョンを使用していることを確認します。

  3. Openflowを設定した ことを確認します。

  4. データベース管理者として、以下のタスクを実行します。

    1. バイナリログ を有効にし、その形式を以下のように保存して構成します。

      log_bin

      on に設定。

      これにより、構造やデータの変更を記録するバイナリログが有効になります。

      binlog_format

      row に設定。

      コネクタは行ベースの複製のみをサポートします。MySQL 8.xバージョンがこの設定をサポートする最後のバージョンとなる可能性があり、将来のバージョンでは行ベースの複製のみがサポートされます。

      GCP クラウド SQL では適用されません。正しい値に固定されています。

      binlog_row_metadata

      full に設定。

      コネクタが動作するためには、すべての行メタデータが必要で、最も重要であるのは列名と主キー情報です。

      binlog_row_image

      full に設定。

      コネクタは、すべての列がバイナリログに書き込まれていることを要求します。

      Amazon Auroraでは適用されません。正しい値に固定されています。

      binlog_row_value_options

      空のままにする。

      このオプションは JSON 列にのみ影響し、 UPDATE ステートメントに対して、 JSON ドキュメントの変更部分のみを含めるように設定できます。コネクタは、完全なドキュメントがバイナリログに書き込まれていることを要求します。

      binlog_expire_logs_seconds

      データベースエージェントが長時間の停止やダウンタイム後も増分複製を継続できるように、少なくとも数時間、またはそれ以上に設定します。Snowflakeでは、コネクタを安定して動作させるために、 バイナリログの有効期限(binlog_expire_logs_seconds) を少なくとも数時間に設定することを推奨します。バイナリログの有効期限終了後に、バイナリログファイルが自動的に削除される場合があります。メンテナンス作業などで統合が長期間中断され、その間に期限切れのバイナリログファイルが削除されると、Openflowはこれらのファイルのデータを複製できなくなります。

      スケジュール複製を使用している場合は、構成されたスケジュールよりも長い値を指定する必要があります。

      以下のコード例をご参照ください。

      log_bin = on
      binlog_format = row
      binlog_row_metadata = full
      binlog_row_image = full
      binlog_row_value_options =
      
      Copy
    2. sort_buffer_size の値を増やします。

      sort_buffer_size = 4194304
      
      Copy

      sort_buffer_size は、 ORDER BY のようなメモリ内ソート操作のためにクエリスレッドごとに割り当てられるメモリ量(バイト単位)を定義します。値が小さすぎると、以下のエラーメッセージが表示されてコネクタが失敗することがあります。

      Out of sort memory, consider increasing server sort buffer size。これは、 sort_buffer_size を増やす必要があることを示します。

    3. Amazon RDS データベースを使用している場合は、 rds_set_configuration を使用して binlog_expire_logs_seconds に関連する保持期間を増やしてください。例えば、binlogを24時間格納したい場合、 mysql.rds_set_configuration('binlog retention hours', 24) を呼び出します。

    4. SSL 経由で接続します。MySQL への SSL 接続を使用する場合は、データベースサーバーのルート証明書を準備します。これは、構成時に必要です。

    5. コネクタのユーザーを作成します。コネクタは、バイナリログを読み取るために REPLICATION_SLAVE と REPLICATION_CLIENT の権限を持つユーザーを必要とします。これらの権限を付与します。

      GRANT REPLICATION SLAVE ON *.* TO '<username>'@'%'
      GRANT REPLICATION CLIENT ON *.* TO '<username>'@'%'
      
      Copy
    6. 複製されたすべてのテーブルに SELECT 権限を付与します。

      GRANT SELECT ON <schema>.* TO '<username>'@'%'
      GRANT SELECT ON <schema>.<table> TO '<username>'@'%'
      
      Copy

      複製セキュリティの詳細については、 バイナリログ をご参照ください。

  5. Snowflakeアカウント管理者として、以下のタスクを実行します。

    1. タイプを SERVICE としてSnowflakeユーザーを作成します。複製データを格納するデータベースを作成し、Snowflakeユーザーに USAGE および CREATE SCHEMA 権限 を付与して、そのデータベースにオブジェクトを作成する権限を設定します。

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
           WITH
               WAREHOUSE_SIZE = 'MEDIUM'
               AUTO_SUSPEND = 300
               AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      
      Copy
    2. セキュアキーのペア(公開キーと秘密キー)を作成します。ユーザーの秘密キーをファイルに格納して、コネクタの構成に提供します。Snowflakeサービスユーザーに公開キーを割り当てます:

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      
      Copy

      詳細については、 キーのペア をご参照ください。

    3. コネクタが使用するウェアハウスを指定します。まずは MEDIUM のウェアハウスサイズから始め、複製するテーブルの量や転送するデータ量に応じてサイズを試してみてください。テーブル数が大きい場合は、通常、ウェアハウスのサイズよりも、 マルチクラスターウェアハウス を使用した方がスケーリングが向上します。

コネクタを設定する

データエンジニアとして、以下のタスクを実行してコネクタをインストールおよび構成します。

コネクタをインストールする

  1. Openflowの概要ページに移動します。 Featured connectors セクションで、 View more connectors を選択します。

  2. Openflowのコネクタページでコネクタを探し、 Add to runtime を選択します。

  3. Select runtime ダイアログで、 Available runtimes ドロップダウンリストからランタイムを選択します。

  4. Add を選択します。

    注釈

    コネクタをインストールする前に、コネクタが取り込んだデータを格納するためのデータベースとスキーマをSnowflakeで作成したことを確認します。

  5. Snowflakeアカウント認証情報でデプロイメントを認証し、Snowflakeアカウントへのランタイムアプリケーションのアクセスを許可するよう求められたられたら、 Allow を選択します。コネクタのインストールプロセスは数分で完了します。

  6. Snowflakeアカウント認証情報でランタイムを認証します。

コネクタプロセスグループが追加されたOpenflowキャンバスが表示されます。

コネクタを構成する

以下のユースケースにコネクタを構成できます。

リアルタイムでテーブルのセットを複製する

  1. インポートしたプロセスグループを右クリックし、 Parameters を選択します。

  2. フローパラメーター の説明に従って、必要なパラメーター値を入力します。

フローパラメーター

まず、 MySQL ソースパラメーターコンテキストのパラメーターを設定し、次に MySQL 宛先パラメーターコンテキストを設定します。これが完了したら、コネクタを有効にします。コネクタを MySQL とSnowflakeの両方に接続し、実行を開始する必要があります。ただし、複製するテーブルが構成に明示的に追加されるまでは、コネクタはデータを複製しません。

複製する特定のテーブルを構成するには、 MySQL 取り込みパラメーターコンテキストを編集します。複製パラメーターコンテキストに変更を適用すると、コネクタが構成を取得し、各テーブルの複製ライフサイクルが開始します。

MySQL ソースパラメーターコンテキスト

パラメーター

説明

MySQL 接続 URL

ソースデータベースへの完全な JDBC URL。コネクタは MySQL と互換性のある MariaDB ドライバーを使用し、 jdbc:mariadb のプレフィックスが URL に必要です。SSL が無効の場合、接続 URL では allowPublicKeyRetrieval パラメーターが true に設定されている必要があります。

例:

  • SSL が有効: jdbc:mariadb://example.com:3306

  • SSL が無効: jdbc:mariadb://example.com:3306?allowPublicKeyRetrieval=true

MySQL JDBC ドライバー

MariaDB JDBC ドライバーのjar への絶対パス。コネクタは MySQL と互換性のある MariaDB ドライバーを使用します。MariaDB JDBC ドライバーをアップロードするには、 Reference asset チェックボックスを選択します。

例: /opt/resources/drivers/mariadb-java-client-3.5.2.jar

MySQL ユーザー名

コネクタのユーザー名。

MySQL パスワード

コネクタのパスワード。

MySQL 宛先パラメーターコンテキスト

パラメーター

説明

宛先データベース

データが永続化されるデータベース。既にSnowflakeに存在している必要があります。

Snowflakeアカウント識別子

データが永続化されるSnowflakeアカウント名(形式: [organization-name]- [account-name])。

Snowflake認証ストラテジー

Snowflakeへの認証のストラテジー。可能な値: SPCS でフローを実行している場合は SNOWFLAKE_SESSION_TOKEN、秘密キーを使用してアクセスを設定する場合は KEY_PAIR

Snowflake秘密キー

認証に使用される RSA 秘密キー。RSA キーは、 PKCS8 標準に従って書式設定され、標準的な PEM ヘッダーとフッターを持つ必要があります。Snowflake秘密キーファイルまたはSnowflake秘密キーのいずれかを定義する必要があることに注意してください。

Snowflake秘密キーファイル

Snowflakeへの認証に使用される RSA 秘密キーを含むファイル。 PKCS8 標準に従って書式設定され、標準的な PEM ヘッダーとフッターを持ちます。ヘッダー行は -----BEGIN PRIVATE で始まります。 Reference asset チェックボックスを選択し、秘密キーファイルをアップロードします。

Snowflake秘密キーパスワード

Snowflake秘密キーファイルに関連付けられたパスワード

Snowflakeロール

クエリ実行時に使用されるSnowflakeロール

Snowflakeのユーザー名

Snowflakeインスタンスへの接続に使用するユーザー名

Snowflakeウェアハウス

クエリの実行に使用されるSnowflakeウェアハウス

MySQL 取り込みパラメーターコンテキスト

パラメーター

説明

含まれるテーブル名

スキーマを含むテーブルパスのコンマ区切りリスト。例: public.my_table, other_schema.other_table

含まれるテーブル正規表現

テーブルパスに一致させる正規表現。式に一致するすべてのパスが複製され、後から作成されたパターンに一致する新しいテーブルも自動的に含まれます。例: public\.auto_.*

フィルター JSON

完全修飾されたテーブル名のリストと、複製に含めるべき列名の正規表現パターンを含む JSON。例: [ {"schema":"public", "table":"table1", "includedPattern":".*name"} ]public スキーマから table1name で終わるすべての列を含めます。

タスクスケジュール CRON をマージする

ジャーナルから宛先テーブルへのマージ操作がトリガーされる期間を定義する CRON 式。連続的なマージやタイムスケジュールでウェアハウスの実行時間を制限したい場合は、 * * * * * ? に設定します。例えば、 * 0 * * * ? という文字列は、丸1時間で1分間のマージをスケジュールしたいことを示しています。 * 20 14 ? * MON-FRI という文字列は、毎週月曜日から金曜日の2:20 PM にマージをスケジュールしたいことを示しています。詳細と例については、 CronTrigger チュートリアル をご参照ください。

テーブルを削除し、複製に再追加する

複製からテーブルを削除するには、複製パラメーターコンテキストの Included Table Names または Included Table Regex パラメーターからテーブルが削除されていることを確認します。

後でテーブルを複製に再追加する場合は、まずSnowflakeで対応する宛先テーブルを削除します。その後、 Included Table Names または Included Table Regex パラメーターにテーブルを追加して戻します。これにより、テーブルの複製プロセスが新しく開始されます。

このアプローチは、失敗したテーブル複製シナリオからの復旧にも使用できます。

テーブルの列のサブセットを複製します。

コネクタは、テーブルごとに複製されるデータを構成列のサブセットにフィルターできます。

列にフィルターを適用するには、複製パラメーターコンテキストのColumn Filterプロパティを変更し、フィルターを適用したいテーブルごとに1エントリずつ、構成の配列を追加します。

列は、名前またはパターン別に包含したり、除外したりすることができます。テーブルごとに単一の条件を適用することも、複数の条件を組み合わせて適用することもできます。除外は常に包含より優先されます。

以下の例は、利用可能なフィールドを示しています。 schema および table フィールドは必須です。 includedexcludedincludedPatternexcludedPattern のうち1つ以上が必要です。

[
    {
        "schema": "<source table schema>",
        "table" : "<source table name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]
Copy

テーブルでデータ変更を追跡する

コネクタはソーステーブルのデータの現在の状態だけでなく、すべての変更セットのすべての行の状態も複製します。このデータは、宛先テーブルと同じスキーマで作成されたジャーナルテーブルに格納されます。

ジャーナルテーブル名の形式は次のとおりです: <ソーステーブル名>_JOURNAL_<タイムスタンプ>_<スキーマ生成>

ここで、 <タイムスタンプ> はソーステーブルが複製に追加されたときのエポック秒の値で、 <スキーマ生成> はソーステーブルのスキーマが変更されるたびに増加する整数です。つまり、スキーマが変更されたソーステーブルは、複数のジャーナルテーブルを持つことになります。

テーブルが複製から削除され、その後再び追加されると、 <タイムスタンプ> の値が変更され、 <スキーマ生成>1 から再び開始されます。

重要

Snowflakeでは、ジャーナルテーブルやその中のデータを一切変更しないことを推奨しています。これらは、複製プロセスの一環として、コネクタが宛先テーブルを更新するために使用されます。

コネクタはジャーナルテーブルをドロップすることはありませんが、複製されたすべてのソーステーブルに対して最新のジャーナルのみをアクティブに使用します。ストレージを再利用したい場合は、複製から削除されたソーステーブルに関連するジャーナルテーブルと、アクティブに複製されたテーブルの最新の世代を除くすべてのジャーナルテーブルを安全にドロップできます。

例えば、コネクタがソーステーブル orders をアクティブに複製するように設定されており、以前にテーブル customers を複製から削除した場合、以下のようなジャーナルテーブルが存在する可能性があります。この場合、 orders_5678_2除いて、それらのすべてをドロップできます。

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

マージタスクのスケジュールを構成する

コネクタはウェアハウスを使用して、変更データキャプチャ(CDC)データを宛先テーブルにマージします。この操作は、 MergeSnowflakeJournalTable プロセッサーによってトリガーされます。新しい変更がない場合、または MergeSnowflakeJournalTable キューで待機する新しいフローファイルがない場合、マージはトリガーされず、ウェアハウスは自動サスペンドします。

ウェアハウスのコストを制限し、スケジュールされた時間のみにマージを制限するには、Merge task Schedule CRON パラメーターで CRON 式を使用します。MergeSnowflakeJournalTable プロセッサーに送られてくるフローファイルをスロットルし、マージは専用の期間のみにトリガーされます。スケジュールに関する詳細は、 スケジュールストラテジー をご参照ください。

フローを実行する

  1. プレーンを右クリックし、 Enable all Controller Services を選択します。

  2. インポートしたプロセスグループを右クリックし、 Start を選択します。コネクタがデータの取り込みを開始します。