の設定 Openflow Connector for MySQL¶

前提条件¶

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

2. Openflowの設定 - BYOC または Openflowの設定 - Snowflakeデプロイメント - タスク概要 があることを確認してください。

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

4. 推奨:ランタイムごとにコネクタインスタンスを1つだけ追加するようにします。

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

   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. 読み取りレプリカを使用して接続する場合は、レプリカでバイナリログを有効にする必要があります。

      構成の詳細は、ステップ4にあります。

   5. バイナリログを有効にした後、ソースから受信したイベントを独自のバイナリログに記録するレプリカを構成します。

      log_replica_updates = ON
      

      Copy

      log_replica_updates は、レプリカがそのソースから受信したイベントを独自のバイナリログに書き込み、それらの変更をレプリカから複製するすべてのデータベースで利用できるようにします。

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

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

      GRANT REPLICATION SLAVE ON *.* TO '<username>'@'%'
      GRANT REPLICATION CLIENT ON *.* TO '<username>'@'%'
      

      Copy

   8. 複製されたすべてのテーブルに SELECT 権限を付与します。

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

      Copy

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

6. 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 ウェアハウスのサイズから、レプリケートされるテーブルの量とデータの転送量に応じてサイズを実験します。テーブル数が大きい場合は、通常、ウェアハウスのサイズよりも、 マルチクラスターウェアハウス を使用した方がスケーリングが向上します。

コネクタを設定する¶

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

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

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

フローを実行する¶

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

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