DataWorks:MySQL

リアルタイム同期

• バージョン 5.6.x より前の MySQL 読み取り専用インスタンスからのデータ同期はサポートされていません。

• 関数インデックスを含むテーブルの同期はサポートされていません。

• XA ROLLBACK はサポートされていません。

  リアルタイム同期は、XA PREPARE 状態のデータを送信先に書き込みますが、XA ROLLBACK が発生した場合、準備されたデータはロールバックされません。このシナリオに対処するには、影響を受けるテーブルをリアルタイム同期タスクから手動で削除し、再度追加して同期を再開する必要があります。

• MySQL サーバーのバイナリログ (binlog) の設定では、ROW フォーマットを使用する必要があります。

• リアルタイム同期では、カスケード削除操作によって削除された関連テーブルのレコードは同期されません。

• Amazon Aurora MySQL データベースの場合、AWS は Aurora MySQL 読み取り専用レプリカで binlog を有効にすることを許可していないため、プライマリ/ライターデータベースに接続する必要があります。リアルタイム同期タスクでは、増分更新を実行するために binlog が必要です。

• オンライン DDL 変更の場合、リアルタイム同期は、Data Management (DMS) を使用して MySQL テーブルに列を追加 (Add Column) することにのみ対応しています。

• MySQL からのストアドプロシージャの読み取りはサポートされていません。

オフライン同期

• MySQL Reader プラグインを使用して複数のテーブルを同期する場合 (たとえば、データベースシャーディングのシナリオ)、同時実行タスクの数がテーブルの数を超えた場合にのみ、システムはテーブルを分割して並列処理を行います。それ以外の場合、システムはテーブルごとに 1 つのタスクを作成します。

• MySQL からのストアドプロシージャの読み取りはサポートされていません。

MySQL バージョンの確認

データ統合は特定の MySQL バージョンをサポートしています。詳細については、「サポートされている MySQL のバージョン」セクションを参照して、ご利用のバージョンがサポートされていることを確認してください。次のステートメントを実行して、MySQL データベースのバージョンを確認します。

SELECT version();

アカウント権限の設定

DataWorks がデータソースにアクセスするために、専用の MySQL アカウントを作成することを推奨します。

1. 任意：アカウントの作成

   詳細については、「MySQL アカウントの作成」をご参照ください。

2. 権限の設定

   • バッチ同期

     • MySQL からデータを読み取るには、アカウントに同期対象テーブルに対する SELECT 権限が必要です。

     • MySQL にデータを書き込むには、アカウントに同期対象テーブルに対する INSERT、DELETE、および UPDATE 権限が必要です。

   • リアルタイム同期

     アカウントには、データベースに対する SELECT、REPLICATION SLAVE、および REPLICATION CLIENT 権限が必要です。

   次のコマンドを実行して、必要な権限を付与します。または、アカウントに SUPER 権限を付与することもできます。次のステートメントでは、'sync_account' をご利用の同期アカウントに置き換えてください。

   -- CREATE USER 'sync_account'@'%' IDENTIFIED BY ''; // 同期アカウントを作成し、パスワードを設定して、任意のホストからのログインを許可します。'%' ワイルドカードは任意のホストを表します。
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'sync_account'@'%'; // 同期アカウントに SELECT、REPLICATION SLAVE、REPLICATION CLIENT 権限を付与します。

   *.* 構文は、すべてのデータベースにあるすべてのテーブルに、指定された権限を付与します。 test データベースの user テーブルなど、特定のテーブルに権限を付与するには、次の文を実行します： GRANT SELECT, REPLICATION CLIENT ON test.user TO 'sync_account'@'%';。

   説明

   REPLICATION SLAVE 権限はグローバル権限であり、特定のデータベースやテーブルに付与することはできません。

MySQL Binlog の有効化 (リアルタイム同期のみ)

データ統合は、MySQL Binlog をサブスクライブすることで、増分データをリアルタイムで同期します。リアルタイム同期タスクを設定する前に、Binlog サービスを有効にする必要があります。

1. Binlog が有効になっているか確認します。

   • 次のステートメントを実行して、Binlog が有効になっているか確認します。

     SHOW variables LIKE "log_bin";

     戻り値が ON の場合、Binlog が有効であることを示します。

   • データ同期にスタンバイデータベースを使用している場合は、次のステートメントを実行して Binlog が有効になっているか確認します。

     SHOW variables LIKE "log_slave_updates";

     戻り値が ON の場合、スタンバイデータベースで Binlog が有効であることを示します。

   返された結果が上記の結果と一致しない場合：

   • セルフマネージド MySQL の場合は、「MySQL 公式ドキュメント」を参照して Binlog を有効にしてください。

   • RDS for MySQL の場合は、「RDS for MySQL のログバックアップ」を参照して Binlog を有効にしてください。

   • PolarDB for MySQL の場合は、「Binlog の有効化」を参照して Binlog を有効にしてください。

2. Binlog のフォーマットを確認します。

   次のステートメントを実行して、Binlog のフォーマットを確認します。

   SHOW variables LIKE "binlog_format";

   戻り値は次のいずれかになります：

   • ROW：Binlog のフォーマットは ROW です。

   • STATEMENT：Binlog のフォーマットは STATEMENT です。

   • MIXED：Binlog のフォーマットは MIXED です。

   重要

   DataWorks のリアルタイム同期は、ROW Binlog フォーマットの MySQL サーバーのみをサポートします。戻り値が ROW でない場合は、Binlog のフォーマットを変更する必要があります。

3. Binlog の完全行ロギングが有効になっているか確認します。

   次のステートメントを実行して、Binlog の完全行ロギングが有効になっているか確認します。

   SHOW variables LIKE "binlog_row_image";

   戻り値は次のいずれかになります：

   • FULL：Binlog の完全行ロギングが有効になっています。

   • MINIMAL：最小行ロギングが有効で、完全行ロギングは有効になっていません。

   重要

   DataWorks のリアルタイム同期は、Binlog の完全行ロギングが有効になっている MySQL サーバーのみをサポートします。戻り値が FULL でない場合は、binlog_row_image の設定を変更する必要があります。

Alibaba Cloud インスタンスモード

MySQL データベースが Alibaba Cloud RDS インスタンスで実行されている場合は、Alibaba Cloud インスタンスモードでデータソースを作成することを推奨します。次の表にパラメーターを示します。

接続文字列モード

より柔軟性のある接続文字列モードでデータソースを作成することもできます。このモードのパラメーターを次の表に示します。

コードエディタを使用したバッチ同期タスクの設定

コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプト形式の要件に基づいて、スクリプト内の関連パラメーターを設定する必要があります。詳細については、「コードエディタの使用」をご参照ください。以下の情報は、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明しています。

Reader スクリプトのデモ

このトピックでは、1 つのデータベース内の単一テーブルとシャーディングの設定例を示します：

• 単一データベース内の単一テーブル

  {
    "type": "job",
    "version": "2.0",//バージョン番号。
    "steps": [
      {
        "stepType": "mysql",//プラグイン名。
        "parameter": {
          "column": [//列名。
            "id"
          ],
          "connection": [
            {
              "querySql": [
                "select a,b from join1 c join join2 d on c.id = d.id;"
              ],
              "datasource": ""//データソース名。
            }
          ],
          "where": "",//フィルター条件。
          "splitPk": "",//シャードキー。
          "encoding": "UTF-8"//エンコーディング形式。
        },
        "name": "Reader",
        "category": "reader"
      },
      {
        "stepType": "stream",
        "parameter": {},
        "name": "Writer",
        "category": "writer"
      }
    ],
    "setting": {
      "errorLimit": {
        "record": "0"//エラーレコード数。
      },
      "speed": {
        "throttle": true,//throttle が false に設定されている場合、mbps パラメーターは効果がなく、速度制限は無効になります。throttle が true に設定されている場合、速度制限は有効になります。
        "concurrent": 1,//ジョブの同時実行数。
        "mbps": "12"//速度制限レート。1 mbps = 1 MB/s。
      }
    },
    "order": {
      "hops": [
        {
          "from": "Reader",
          "to": "Writer"
        }
      ]
    }
  }

• シャーディング

  説明

  MySQL Reader はシャーディングをサポートしており、異なるデータソースにまたがる同じスキーマを持つ複数のテーブルからデータを読み取り、単一の送信先テーブルに書き込むことができます。データベース全体のシャーディングを設定するには、データ統合でタスクを作成し、全データベースシャーディング機能を選択する必要があります。

  {
    "type": "job",
    "version": "2.0",
    "steps": [
      {
        "stepType": "mysql",
        "parameter": {
          "indexes": [
            {
              "type": "unique",
              "column": [
                "id"
              ]
            }
          ],
          "envType": 0,
          "useSpecialSecret": false,
          "column": [
            "id",
            "buyer_name",
            "seller_name",
            "item_id",
            "city",
            "zone"
          ],
          "tableComment": "テスト注文テーブル",
          "connection": [
            {
              "datasource": "rds_dataservice",
              "table": [
                "rds_table"
              ]
            },
            {
              "datasource": "rds_workshop_log",
              "table": [
                "rds_table"
              ]
            }
          ],
          "where": "",
          "splitPk": "id",
          "encoding": "UTF-8"
        },
        "name": "Reader",
        "category": "reader"
      },
      {
        "stepType": "odps",
        "parameter": {},
        "name": "Writer",
        "category": "writer"
      },
      {
        "name": "Processor",
        "stepType": null,
        "category": "processor",
        "copies": 1,
        "parameter": {
          "nodes": [],
          "edges": [],
          "groups": [],
          "version": "2.0"
        }
      }
    ],
    "setting": {
      "executeMode": null,
      "errorLimit": {
        "record": ""
      },
      "speed": {
        "concurrent": 2,
        "throttle": false
      }
    },
    "order": {
      "hops": [
        {
          "from": "Reader",
          "to": "Writer"
        }
      ]
    }
  }

Reader スクリプトのパラメーター

Writer スクリプトのデモ

{
  "type": "job",
  "version": "2.0",//バージョン番号。
  "steps": [
    {
      "stepType": "stream",
      "parameter": {},
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "mysql",//プラグイン名。
      "parameter": {
        "postSql": [],//実行後 SQL ステートメント。
        "datasource": "",//データソース。
        "column": [//列名。
          "id",
          "value"
        ],
        "writeMode": "insert",//書き込みモード。insert、replace、または update に設定できます。
        "batchSize": 1024,//バッチサイズ。
        "table": "",//テーブル名。
        "nullMode": "skipNull",//NULL 値の処理ポリシー。
        "skipNullColumn": [//NULL 値がスキップされる列。
          "id",
          "value"
        ],
        "preSql": [
          "delete from XXX;"//実行前 SQL ステートメント。
        ]
      },
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "errorLimit": {//エラーレコード数。
      "record": "0"
    },
    "speed": {
      "throttle": true,//throttle が false に設定されている場合、mbps パラメーターは効果がなく、速度制限は無効になります。throttle が true に設定されている場合、速度制限は有効になります。
      "concurrent": 1,//ジョブの同時実行数。
      "mbps": "12"//速度制限レート。ソース/送信先データベースへの過剰な読み取り/書き込み圧力を防ぐために、最大同期速度を制御します。1 mbps = 1 MB/s。
    }
  },
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  }
}

Writer スクリプトのパラメーター