DataWorks:MySQL データソース

リアルタイム同期

• 5.6.x より前のバージョンを実行している MySQL の読み取り専用インスタンスからはデータを同期できません。

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

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

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

• MySQL サーバーのバイナリログフォーマットは ROW に設定する必要があります。

• リアルタイム同期は、関連テーブルのカスケード削除を同期しません。

• Amazon Aurora MySQL データベースの場合、プライマリ/ライターインスタンスに接続する必要があります。Aurora MySQL の読み取り専用レプリカでは Binlog 機能を有効にできず、リアルタイム同期タスクでは増分更新のために Binlog が必要です。

• MySQL テーブルのオンライン DDL 変更では、リアルタイム同期は Data Management Service (DMS) を介した列の追加 (Add Column) のみをサポートしています。

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

オフライン同期

• MySQL Reader プラグインを使用してシャーディングされたデータベースから複数のテーブルを同期する場合、テーブル分割を有効にするには、同時実行数をテーブル数より大きい値に設定する必要があります。そうしないと、システムはテーブルごとに 1 つのタスクを作成します。

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

MySQL バージョンの確認

Data Integration は特定の MySQL バージョンのみをサポートします。詳細については、「サポートされている MySQL バージョン」セクションをご参照ください。次のコマンドを実行して MySQL のバージョンを確認してください：

SELECT version();

アカウント権限の設定

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

1. オプション：アカウントの作成

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

2. 権限の付与

   • バッチ

     バッチ同期シナリオでは：

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

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

     説明

     MySQL からデータを読み取って同期するだけで、データを書き戻す必要がない場合は、アカウントに SELECT 権限を付与するだけで十分です。INSERT、DELETE、および UPDATE 権限は読み取り専用の同期には必要ありません。

   • リアルタイム

     リアルタイム同期シナリオでは、アカウントにデータベースに対する SELECT、REPLICATION SLAVE、および REPLICATION CLIENT 権限が必要です。

   次のコマンドを実行して、必要な権限を付与できます。または、アカウントに SUPER 権限を付与することもできます。次のコマンドで、'sync_account' を作成したアカウントの名前に置き換えてください。

   -- オプション：任意のホスト ('%') から接続できる同期アカウントを作成します。
   -- CREATE USER 'sync_account'@'%' IDENTIFIED BY 'your_password'; 
   -- 同期アカウントに必要な権限を付与します。
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'sync_account'@'%';

   *.* 構文は、すべてのデータベースのすべてのテーブルに対する権限を付与します。特定のデータベースの特定のテーブルに対する権限を付与することもできます。たとえば、test データベースの user テーブルに権限を付与するには、次のコマンドを実行します：GRANT SELECT, REPLICATION CLIENT ON test.user TO 'sync_account'@'%';。

   説明

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

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

Data Integration は、MySQL の binlog をサブスクライブすることで、リアルタイムの増分データ同期を実行します。DataWorks で同期タスクを設定する前に、MySQL の binlog を有効にする必要があります。

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

   • 次の文を実行して、binlog が有効になっているか確認します。

     SHOW variables LIKE "log_bin";

     返された値が ON の場合、binlog は有効です。

   • レプリカデータベースの binlog ステータスを確認するには、次の文を実行します。

     SHOW variables LIKE "log_slave_updates";

     返された値が ON の場合、レプリカデータベースで binlog が有効になっています。

   返された値が期待どおりでない場合：

   • オープンソースの MySQL の場合、binlog を有効にする手順については、MySQL 公式ドキュメントをご参照ください。

   • RDS for MySQL の場合、binlog を有効にする手順については、「ApsaraDB RDS for MySQL インスタンスのバックアップ」をご参照ください。

   • PolarDB for MySQL の場合、binlog を有効にする手順については、「バイナリロギングの有効化または無効化」をご参照ください。

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

   次の文を実行して、binlog フォーマットを確認します。

   SHOW variables LIKE "binlog_format";

   考えられる戻り値：

   • 戻り値が ROW の場合、有効な binlog フォーマットは ROW です。

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

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

   重要

   DataWorks のリアルタイム同期は ROW フォーマットのみをサポートします。返された値が ROW でない場合は、binlog_format の設定を変更する必要があります。

3. 完全な行イメージがログに記録されているか確認します。

   行イメージの設定を確認するには、以下の文を実行します。

   SHOW variables LIKE "binlog_row_image";

   返される値：

   • FULL：完全な行イメージがログに記録されます。

   • MINIMAL：完全な行イメージの代わりに、最小限の行イメージがログに記録されます。

   重要

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

Alibaba Cloud インスタンスモード

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

接続文字列モード

より柔軟な設定のために、接続文字列モードを使用してデータソースを作成することもできます。パラメーターを次のように設定します：

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

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

Reader スクリプトの例

以下の例は、単一データベース内の単一テーブルと、シャーディングされたテーブルの設定を示しています。

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

  {
    "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,// レート制限を有効にする場合は true に設定します。false の場合、レート制限は無効になり、'mbps' パラメーターは無視されます。
        "concurrent": 1,// 並行スレッド数。
        "mbps": "12"// レート制限。1 mbps = 1 MB/s。
      }
    },
    "order": {
      "hops": [
        {
          "from": "Reader",
          "to": "Writer"
        }
      ]
    }
  }

• シャーディングされたテーブル

  説明

  MySQL Reader は、同じスキーマを持つ複数の MySQL テーブルからデータを読み取ることができます。この文脈での「シャーディングされたテーブル」とは、複数のソーステーブルのデータが単一の送信先テーブルに書き込まれるシナリオを指します。データベースレベルでシャーディングされたテーブルを同期するには、Data Integration でタスクを作成し、データベース全体のシャーディング機能を選択します。

  {
    "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,// レート制限を有効にする場合は true に設定します。false の場合、レート制限は無効になり、'mbps' パラメーターは無視されます。
      "concurrent": 1,// 並行スレッド数。
      "mbps": "12"// データベースへの過剰な I/O 圧力を防ぐための MB/s 単位のレート制限。1 mbps = 1 MB/s。
    }
  },
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  }
}

Writer パラメーター