DataWorks:MySQL データソース

リアルタイム読み取り

• MySQL 読み取り専用インスタンスからのデータ同期はサポートされていません。

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

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

  XA PREPARE 状態のトランザクション内のデータについては、リアルタイム同期がデータを宛先に書き込みます。XA ROLLBACK が発生した場合、リアルタイム同期はこのデータをロールバックしません。このシナリオを処理するには、リアルタイム同期タスクからテーブルを手動で削除し、再度追加して同期を再開する必要があります。

• MySQL サーバーのバイナリロギングでは、ROW フォーマットのみがサポートされています。

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

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

• オンライン DDL 操作のリアルタイム同期は、Data Management DMS を使用した MySQL テーブルへの列の追加 (Add Column) のみをサポートします。

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

オフライン読み取り

• MySQL Reader プラグインがシャード化されたデータベース内の複数のテーブルからデータを同期する場合、デフォルトではシャード化されたタスクの数はテーブルの数と等しくなります。単一のテーブルでシャーディングを有効にするには、同時タスクの数をテーブルの数より大きい値に設定する必要があります。

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

MySQL バージョンの確認

Data Integration には 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 'password'; //同期アカウントを作成し、パスワードを設定し、任意のホストからデータベースにログインできるようにします。パーセント記号 (%) は任意のホストを示します。
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'sync_account'@'%'; // データベースに対する SELECT、REPLICATION SLAVE、および REPLICATION CLIENT 権限を同期アカウントに付与します。

   *.* は、すべてデータベースのすべてのテーブルに対する指定された権限をアカウントに付与します。ターゲットデータベースの特定のテーブルに権限を付与することもできます。たとえば、test データベースの user テーブルに `SELECT` および `REPLICATION CLIENT` 権限を付与するには、次の文を実行します: GRANT SELECT, REPLICATION CLIENT ON test.user TO 'sync_account'@'%';。

   説明

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

MySQL バイナリロギングを有効にする (リアルタイム同期のみ)

Data Integration は、MySQL バイナリログをサブスクライブすることで、増分データのリアルタイム同期を実行します。DataWorks で同期タスクを構成する前に、MySQL バイナリロギングサービスを有効にする必要があります。手順は次のとおりです:

1. バイナリロギングが有効になっているかどうかを確認します。

   • 次の文を実行して、バイナリロギングが有効になっているかどうかを確認します。

     SHOW variables LIKE "log_bin";

     コマンドが ON を返した場合、バイナリロギングは有効になっています。

   • セカンダリデータベースからデータを同期する場合、次の文を実行してバイナリロギングが有効になっているかどうかを確認することもできます。

     SHOW variables LIKE "log_slave_updates";

     コマンドが ON を返した場合、セカンダリデータベースでバイナリロギングが有効になっています。

   結果が一致しない場合:

   • オープンソースの MySQL の場合、MySQL の公式ドキュメントを参照してバイナリロギングを有効にしてください。

   • ApsaraDB RDS for MySQL の場合、RDS for MySQL のログバックアップを参照してバイナリロギングを有効にしてください。

   • PolarDB for MySQL の場合、バイナリロギングを有効にするを参照してバイナリロギングを有効にしてください。

2. バイナリログのフォーマットを確認します。

   次の文を実行して、バイナリログのフォーマットを確認します。

   SHOW variables LIKE "binlog_format";

   考えられる戻り値は次のとおりです:

   • ROW: バイナリログのフォーマットは ROW です。

   • STATEMENT: バイナリログのフォーマットは STATEMENT です。

   • MIXED: バイナリログのフォーマットは MIXED です。

   重要

   DataWorks のリアルタイム同期は、MySQL サーバーのバイナリロギングに対して ROW フォーマットのみをサポートします。コマンドが ROW 以外の値を返す場合は、バイナリログのフォーマットを変更する必要があります。

3. フル行イメージロギングが有効になっているかどうかを確認します。

   次の文を実行して、フル行イメージロギングが有効になっているかどうかを確認します。

   SHOW variables LIKE "binlog_row_image";

   考えられる戻り値は次のとおりです:

   • FULL: バイナリログに対してフル行イメージロギングが有効になっています。

   • MINIMAL: バイナリログに対して最小行イメージロギングが有効になっています。フル行イメージロギングは有効になっていません。

   重要

   DataWorks のリアルタイム同期は、フル行イメージロギングが有効になっている 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,// throttle が false に設定されている場合、mbps パラメーターは効果がなく、データ転送レートは制限されません。throttle が true に設定されている場合、データ転送レートは制限されます。
        "concurrent": 1,// 同時ジョブの数。
        "mbps": "12"// 最大データ転送レート。1 mbps = 1 MB/s。
      }
    },
    "order": {
      "hops": [
        {
          "from": "Reader",
          "to": "Writer"
        }
      ]
    }
  }

• シャード化されたテーブル

  説明

  シャーディングにより、MySQL Reader は同じスキーマを持つ複数の MySQL テーブルを選択できます。この文脈では、シャーディングとは、複数のソース 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": "Test order table",
          "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,// 1 つのバッチで送信されるレコードの数。
        "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 スクリプトパラメーター