DataWorks:MySQL データソース

リアルタイム読み取り

• バージョン 5.6.x より前の 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": "テスト注文テーブル",
          "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 スクリプトのパラメーター