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

リアルタイム同期タスクを使用すると、MySQL データベース全体の増分データを LogHub (SLS) にリアルタイムで同期できます。このトピックでは、MySQL データベース全体を LogHub に同期するためのリアルタイム同期タスクの作成方法と、作成後のタスクステータスの表示方法について説明します。

前提条件

Serverless リソースグループまたはデータ統合専用リソースグループを購入済みであること。
MySQL データソースと LogHub (SLS) データソースを作成済みであること。詳細については、「データソースの構成」をご参照ください。
リソースグループとデータソース間のネットワーク接続が確立されていること。詳細については、「ネットワーク接続ソリューションの概要」をご参照ください。

操作手順

1. 同期タスクタイプの選択

Data Integration ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ統合] > [データ統合] を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ統合へ] をクリックします。
左側のナビゲーションウィンドウで [同期タスク] をクリックします。ページの上部で [新しい同期タスク] をクリックします。表示されるダイアログボックスで、次の基本パラメーターを設定します。
- [ソースタイプ]： MySQL。
- [宛先タイプ]： LogHub。
- [タスク名]：同期タスクのカスタム名。
- [タスクタイプ]： データベース全体のリアルタイム移行。

2. ネットワークとリソースの構成

[ネットワークとリソースの構成] セクションで、同期タスクの [リソースグループ] を選択します。[タスクリソース使用量] で CU を割り当てることができます。
[ソース] を MySQL データソースに、[宛先] を LogHub データソースに設定します。次に、[接続テスト] をクリックします。
ソースと宛先の両方のデータソースで接続テストに合格したら、[次へ] をクリックします。

3. データ同期のためのデータソースとテーブルの選択

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

[指定したデータベースとテーブルを選択]：
- [ソースデータベースとテーブル] セクションの [データベースのフィルタリング] および [テーブルの検索] フィールドにキーワードを入力して、同期するデータベースとテーブルをフィルタリングします。必要なデータベースとテーブルを選択し、[] アイコンをクリックして [選択したテーブル] セクションに移動します。
- [選択したテーブル] エリアで、[データベースのフィルタリング] および [テーブルの検索] フィールドを使用して、同期から除外するデータベースとテーブルを検索します。これらの項目を選択し、[] アイコンをクリックして、[ソーステーブル] エリアに移動します。
[正規表現を使用 (正規表現に基づく実行中のテーブルの追加と削除をサポート)]：
[データベースのフィルタリング] および [テーブルのフィルタリング] フィールドに正規表現を入力して、データベースとテーブルをフィルタリングします。[OK] をクリックして、同期するデータベースとテーブルを選択します。
説明
たとえば、データベースをプレフィックス a で、テーブルをプレフィックス order でフィルターするには、[データベースのフィルタリング] ボックスに a.* を、[テーブルのフィルタリング] ボックスに order.* を入力します。

4. 宛先テーブルのマッピング

データを同期するテーブルを選択すると、選択したテーブルが [宛先テーブルのマッピングルール] セクションに自動的に表示されます。宛先テーブルのプロパティはマッピング待ちの状態です。データの読み取りと書き込みの関係を決定するには、ソーステーブルと宛先テーブル間のマッピングを手動で定義する必要があります。その後、[操作] 列の [リフレッシュ] をクリックします。ソーステーブルと宛先テーブル間のマッピングを直接リフレッシュできます。宛先テーブルに関連する設定を構成した後で、ソーステーブルと宛先テーブル間のマッピングをリフレッシュすることもできます。

1. 宛先 Logstore の選択

[宛先 Logstore] 列のドロップダウンリストから、同期タスクの宛先 Logstore を選択します。一度に複数のテーブルを構成するには、それらを選択し、[一括変更] > [宛先 Logstore] をクリックします。

2. DML ルールの設定

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

単一のテーブルのルールを定義するには、[DML ルール構成] 列の [構成] をクリックします。
ルールを一括で定義する：同期するすべてのテーブルを選択します。リストの下部で、[一括変更] > [DML ルール構成] を選択します。

5. アラートルールの設定

同期タスクの失敗によるビジネスデータの同期遅延を防ぐために、同期タスクに異なるアラートルールを設定できます。

ページの右上隅にある [アラートルールの設定] をクリックして、[リアルタイム同期サブノードのアラートルール構成] パネルに移動します。
[アラートルールの設定] パネルで、[アラートルールの追加] をクリックします。[アラートルールの追加] ダイアログボックスで、パラメーターを設定してアラートルールを構成します。
説明
このステップで構成するアラートルールは、同期タスクによって生成されるリアルタイム同期サブタスクに対して有効になります。同期タスクの構成が完了したら、「リアルタイム同期タスクの管理」をご参照のうえ、[リアルタイム同期タスク] ページに移動し、リアルタイム同期サブタスクに構成されたアラートルールを変更できます。
アラートルールの管理。
作成されたアラートルールを有効または無効にできます。アラートの重大度レベルに基づいて、異なるアラート受信者を指定することもできます。

6. 詳細パラメーターの設定

同期タスクを詳細に構成するには、ページの右上隅にある [詳細パラメーターのカスタマイズ] をクリックします。

ページの右上隅にある [詳細パラメーターの設定] をクリックします。

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

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

7. リソースグループの設定

ページの右上隅にある [リソースグループの設定] をクリックして、現在の同期タスクの実行に使用されるリソースグループを表示および変更できます。

8. 同期タスクの実行

同期タスクの構成が完了したら、ページ下部の [完了] をクリックします。
[データ統合] ページの [同期タスク] セクションで、作成した同期タスクを見つけ、[操作] 列の [開始] をクリックします。
[タスク] セクションで同期タスクの [名前/ID] をクリックし、同期タスクの詳細な実行プロセスを表示します。

同期タスクの O&M

タスクステータスの表示

同期タスクを作成した後、[同期タスク] ページで作成されたタスクのリストとその基本情報を表示できます。

[操作] 列では、同期タスクを [開始] または [停止] できます。[その他] をクリックして、タスクを [編集] または [表示] します。
実行中のタスクについては、[実行概要] セクションでその基本ステータスを表示できます。概要の領域をクリックして実行の詳細を表示することもできます。
MySQL から LogHub (SLS) へのリアルタイム同期タスクの場合、リアルタイム同期のパフォーマンス統計が表示されます。これらの統計には、リアルタイムの進捗、DDL レコード、DML レコード、およびアラート情報が含まれます。

同期タスクの再実行

特殊なケースで、ソースにテーブルを追加または削除したり、宛先テーブルのスキーマや名前を変更したりした場合、同期タスクの [操作] 列の [その他] をクリックし、[再実行] をクリックして変更後にタスクを再実行できます。再実行プロセス中、同期タスクは新しく追加されたテーブルから宛先へのデータのみを同期するか、マッピングされたソーステーブルからスキーマまたは名前が変更された宛先テーブルへのデータのみを同期します。

タスクの構成を変更せずに同期タスクを再実行する場合は、[操作] 列の [その他] をクリックし、[再実行] をクリックしてタスクを再実行し、完全同期と増分同期を再度実行します。
タスクにテーブルを追加または削除した後に同期タスクを再実行する場合は、変更後に [完了] をクリックします。この場合、同期タスクの [操作] 列に [更新を適用] が表示されます。[更新を適用] をクリックして、システムに同期タスクを再実行させます。再実行プロセス中、同期タスクは新しく追加されたテーブルから宛先にデータを同期します。元のテーブルのデータは再度同期されません。

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

展開しない

{
  "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"
}