MySQL データベース全体の LogHub (SLS) へのリアルタイム同期 - DataWorks

MySQL から LogHub (SLS) へのリアルタイム同期パイプラインは、増分データを送信先に同期します。このトピックでは、MySQL からデータベース全体を LogHub (SLS) へリアルタイムで同期するタスクを作成し、そのステータスをモニタリングする方法について説明します。

前提条件

サーバーレスリソースグループを購入済みであること。
MySQL データソースと LogHub (SLS) データソースを作成済みであること。詳細については、「データソースの構成」をご参照ください。
リソースグループとデータソースの間にネットワーク接続が存在すること。詳細については、「ネットワーク接続ソリューションの概要」をご参照ください。

操作手順

ステップ 1：同期タスクタイプの選択

DataWorks コンソールにログインします。対象のリージョンで、左側のナビゲーションウィンドウで データ統合 > データ統合 をクリックします。ドロップダウンリストからワークスペースを選択し、入力 データ統合 をクリックします。
左側のナビゲーションウィンドウで、Synchronization Task をクリックします。ページの上部で、Create Synchronization Task をクリックします。ダイアログボックスで、次の基本設定を構成します：
- Source Type： MySQL。
- Destination Type： LogHub。
- Specific Type： データベース全体のリアルタイム同期。
- Synchronization Mode：増分同期。このメソッドは、ビジネスデータベースから LogHub に binlog データを継続的に同期します。

ステップ 2：ネットワークとリソースの構成

Network and Resource Configuration セクションで、同期タスクの Resource Group を選択します。Task Resource Usage のために計算ユニット (CU) を割り当てることができます。
Source Information には、追加した MySQL データソースを選択します。Destination には、追加した LogHub データソースを選択します。次に、Test Connectivity をクリックします。
両方の接続性テストに合格したら、Next をクリックします。

ステップ 3：データベースとテーブルの選択

このステップでは、[ソーステーブル] エリアで同期するソーステーブルを選択し、アイコンをクリックして右側の [選択したテーブル] エリアに移動します。

[ソーステーブル] エリアでは、リストを貼り付けてテーブルを選択 することもできます。最大 5,000 個のテーブルを選択できます。

Select Specific Databases/Tables：
- [ソーステーブル] エリアで、Database Filter と Table Filter フィールドを使用して、同期したいデータベースとテーブルを検索します。項目を選択し、アイコンをクリックして Selected Databases/Tables エリアに移動します。
- Selected Databases/Tables エリアで、Database Filter と Table Filter フィールドを使用して、除外したいテーブルを検索します。それらを選択し、アイコンをクリックして Source Databases/Tables エリアに戻します。
Use regex for table selection (supports adding/removing tables by regex during runtime).：

Database Filter と Table Filter フィールドに正規表現を入力します。Confirm をクリックして、一致するテーブルを選択します。

説明
例えば、名前が a で始まるデータベースのテーブルと、名前が order で始まるテーブルを同期するには、Database Filter ボックスに a.* を、Table Filter ボックスに order.* を入力します。

ステップ 4：送信先テーブルのマッピング

同期するテーブルを選択すると、このページに自動的に表示されます。ただし、送信先テーブルはまだマッピングされていません。データの読み取りと書き込み方法を指定する、ソーステーブルと送信先テーブル間のマッピングを定義する必要があります。マッピングを定義した後、Refresh Mapping をクリックして次のステップに進みます。マッピングは今すぐ更新することも、送信先テーブルのルールをカスタマイズした後に更新することもできます。

1. 送信先 Logstore の選択

Destination Logstore 列で、ドロップダウンリストから同期タスクの送信先 Logstore を選択します。複数のテーブルを選択してから Batch Modify > Destination Logstore をクリックして一括で構成することもできます。

2. DML ルールの構成

Data Integration はデフォルトの DML 処理ルールを提供します。送信先テーブルに書き込む DML コマンドに対して、カスタムの処理ルールを定義することもできます。

単一テーブルのルールを定義する：特定のテーブルの DML Rule Configuration 列にある Configuration をクリックして、そのテーブルの DML ルールを定義します。
ルールを一括で定義する：同期するすべてのテーブルを選択し、リストの下部にある Batch Modify > DML Rule Configuration をクリックします。

ステップ 5：アラートの構成

タスクエラーによるデータ同期の遅延を防ぐために、同期タスクのアラートポリシーを構成できます。

ページ右上の Alert Settings をクリックして、Alert Rule Configurations for Real-time Synchronization Subnode 設定ページに移動します。
Add Alert Rule をクリックして、アラートルールを構成します。

説明
ここで定義されたアラートルールは、このタスクによって生成されるリアルタイム同期サブタスクに適用されます。タスクの構成が完了したら、「リアルタイム同期タスクの実行と管理」ページに移動して、リアルタイム同期サブタスクのモニタリングアラートルールを表示および変更できます。
アラートルールの管理。

既存のアラートルールについては、アラートのトグルを使用してアラートルールを有効または無効にできます。また、アラートの重大度レベルに基づいて、異なる連絡先にアラートを送信することもできます。

ステップ 6：詳細パラメーターの構成

タスクを微調整するには、Custom Advanced Parameters 列の Configuration をクリックして詳細パラメーターを変更します。

ページ右上の [詳細パラメーター] をクリックして、詳細パラメーター構成ページに移動します。

次の表に、主要なパラメーターを示します。

パラメーター	説明
[ビジネスデータの展開]	ソーステーブルのビジネスデータフィールドをログエントリのトップレベルにフラット化するかどうかを指定します。 true：データを展開します。展開フォーマットについては、[データ展開フォーマット] セクションで [部分的な展開] または [完全な展開] を選択できます。 false (デフォルト)：データを展開しません。データは単一の JSON フィールド内にカプセル化されます。
[DDL 情報のパススルー]	`CREATE TABLE` や `ALTER TABLE` などのデータ定義言語 (DDL) 操作をソーステーブルからキャプチャし、ログエントリとして SLS に同期するかどうかを制御します。 true (デフォルト)：DDL 情報をパススルーします。 false：DDL 情報をパススルーしません。
[ソースの NULL 値の処理]	ソースフィールドが NULL の場合に送信先フィールドに書き込む値を指定します。デフォルトでは空で、NULL 値が保持されます。
[データ展開フォーマット]	ビジネスデータ (`data` フィールド) が SLS に書き込まれる際の構造を定義します。このオプションは、[ビジネスデータの展開] オプションが true に設定されている場合にのみ表示されます。選択したフォーマットは、下流システムがログを消費する方法に直接影響します。フォーマットの例については、「付録：データ展開フォーマットの例」をご参照ください。部分的な展開 (デフォルト)：MySQL binlog の Logtail コレクションフォーマットと互換性があります。完全な展開：変更の変更前後のイメージを含むすべてのビジネスデータフィールドを、データ内の独立したトップレベルのキーと値のペアとしてフラット化します。

ステップ 7：リソースグループの構成

ページ右上の Configure Resource Group をクリックして、現在のタスクで使用されているリソースグループを表示および切り替えます。

ステップ 8：同期タスクのデプロイと実行

すべての構成が完了したら、Save をクリックしてタスクを保存します。
Data Integration > Sync Tasks ページで、作成した同期タスクを見つけ、Operation 列の Deploy をクリックします。デプロイ時に Start immediately after deployment を選択した場合、確認後すぐにタスクが実行されます。そうでない場合は、手動でタスクを開始する必要があります。

説明
Data Integration タスクは、実行する前に本番環境にデプロイする必要があります。したがって、タスクを作成または編集した後は、変更を有効にするために Publish 操作を実行する必要があります。
Tasks 内の対応するタスクの Name/ID をクリックして、タスクの詳細な実行プロセスを表示します。

同期タスクの管理

タスクステータスの表示

同期タスクを作成した後、同期タスクページで作成されたタスクのリストとその基本情報を表示できます。タスクが正常に作成されると、タスクリストにステータスが [実行中] として表示されます。実行概要では、リアルタイム同期ステータスは [開始待ち] で、リソースグループは Serverless_Group です。

[操作] 列では、同期タスクを Start または Stop できます。[その他] の下には、Edit や View などの他のアクションがあります。
実行中のタスクについては、Execution Overview を確認してその運用ステータスを確認したり、概要内のエリアをクリックして詳細を表示したりできます。タスク詳細ページの [基本情報] セクションには、タスク ID、データソース、リソースグループ、同期ソリューション、および現在の CU 使用量が表示されます。[実行ステータス]、[実行ログ]、[フェイルオーバー]、および [リソース使用量] タブを切り替えてタスクをモニターできます。

MySQL から LogHub (SLS) へのリアルタイム同期タスクは、進捗、DDL レコード、DML レコード、アラート情報などのリアルタイム統計を表示します。

タスクの再実行

テーブルの追加や削除、送信先テーブルスキーマやテーブル名情報の変更が必要な場合など、特定のシナリオでは、同期タスクの Operations 列にある Rerun をクリックできます。システムは新しく追加または変更されたテーブルのみを同期します。以前に同期されたテーブルや変更されていないテーブルは再同期されません。

Rerun をクリックして、完全な初期化とリアルタイム同期を再実行します。
タスクを編集してテーブルを追加または削除し、タスクを保存してからデプロイします。デプロイ後、[操作] 列に Apply Updates ボタンが表示されます。Apply Updates をクリックして、変更されたタスクの再実行をトリガーします。新しく追加または変更されたテーブルのみが同期されます。以前に同期されたテーブルは再同期されません。

付録：データ展開フォーマットの例

展開なし

{
  "ExecutionTime": 1761017850000,
  "_db_": "*****",
  "_event_": "row_update",
  "_event_time_": 1761017850,
  "_file_name_": "mysql-bin.*****",
  "_gtid_": "4a21a3ce-ad7a-11f0-a8f3**********",
  "_host_": "rm-*********.mysql.rds.aliyuncs.com",
  "_id_": "176101777********",
  "_offset_": "265*****",
  "_table_": "t_parameter",
  "data": {
    "updateBefore": {
      "_old_id": "3",
      "_old_name": "82174b93-b810-4030-8652-e5c1667d3f72",
      "_old_value": "+@}8-/XC",
      "_old_status": "kBdO",
      "_old_description": "a?!L7{jaH+",
      "_old_create_time": "2023-12-28 19:03:43",
      "_old_create_user": "+Zs",
      "_old_modify_time": "2006-11-26 20:42:31",
      "_old_modify_user": "brTYGI?jLL"
    },
    "updateAfter": {
      "id": "3",
      "name": "82174b93-b810-4030-8652-e5c1667d3f72-tagd",
      "value": "+@}8-/XC",
      "status": "kBdO",
      "description": "a?!L7{jaH+",
      "create_time": "2023-12-28 19:03:43",
      "create_user": "+Zs",
      "modify_time": "2006-11-26 20:42:31",
      "modify_user": "brTYGI?jLL"
    }
  }
}

部分的な展開：MySQL binlog の Logtail コレクションフォーマットと互換性あり

{
  "ExecutionTime": 1761017850000,
  "_db_": "*****",
  "_event_": "row_update",
  "_event_time_": 1761017850,
  "_file_name_": "mysql-bin.*****",
  "_gtid_": "4a21a3ce-ad7a-11f0-a8f3**********",
  "_host_": "rm-*********.mysql.rds.aliyuncs.com",
  "_id_": "176101777********",
  "_offset_": "265*****",
  "_table_": "t_parameter",
  "old_data": {
    "_old_id": "1",
    "_old_name": "0e459c1a-c6ce-459b-b374-a161b095c8e9",
    "_old_value": "Hello",
    "_old_status": "b",
    "_old_description": "cw",
    "_old_create_time": "2007-08-06 16:19:03",
    "_old_create_user": "!wW4",
    "_old_modify_time": "2017-04-21 18:21:58",
    "_old_modify_user": "s"
  },
  "data": {
    "id": "1",
    "name": "0e459c1a-c6ce-459b-b374-a161b095c8e9-dsg",
    "value": "Hello",
    "status": "b",
    "description": "cw",
    "create_time": "2007-08-06 16:19:03",
    "create_user": "!wW4",
    "modify_time": "2017-04-21 18:21:58",
    "modify_user": "s"
  }
}

完全な展開：すべてのフィールドがフラット化されます。

{
  "ExecutionTime": 1761017850000,
  "_db_": "****",
  "_event_": "row_update",
  "_event_time_": 1761017850,
  "_file_name_": "mysql-bin.*****",
  "_gtid_": "4a21a3ce-ad7a-11f0-a8f3**********",
  "_host_": "rm-*********.mysql.rds.aliyuncs.com",
  "_id_": "176101777********",
  "_offset_": "265*****",
  "_table_": "t_parameter",
  "_old_create_time": "2024-09-27 15:27:10",
  "_old_create_user": "o",
  "_old_description": "LZ[1HsTE",
  "_old_id": "6",
  "_old_modify_time": "2008-03-15 08:05:53",
  "_old_modify_user": "/{=>7_d@0Q",
  "_old_name": "cf8a671c-4414-45f5-a22c-62c353a6f1ef",
  "_old_status": "K:HQOX-?gK",
  "_old_value": "23]sn<t",
  "create_time": "2024-09-27 15:27:10",
  "create_user": "o",
  "description": "LZ[1HsTE",
  "id": "6",
  "modify_time": "2008-03-15 08:05:53",
  "modify_user": "/{=>7_d@0Q",
  "name": "cf8a671c-4414-45f5-a22c-62c353a6f1efgsdsa",
  "status": "K:HQOX-?gK",
  "value": "23]sn<t"
}