データ同期操作ガイド
ダウンロード
フォーカスモード
フォントサイズ
DTSは、ソースデータベースの全量および増分データをCkafkaに同期することをサポートしており、ユーザーがビジネス変更データを迅速に取得して分析・応用するのに便利です。このドキュメントでは、DTSを使用してMySQL/MariaDB/Percona/TDSQL-C MySQLデータをTencent CloudデータベースCkafkaに同期するプロセスについて説明します。
ソースデータベースがサポートする配置タイプは以下の通りです：
自社構築MySQL、サードパーティクラウドベンダーMySQL、TencentDB for MySQL。
自社構築MariaDB、TencentDB for MariaDB。
自社構築Percona。
TencentDB for TDSQL-C MySQL。
注意事項
同じタスクは複数のTopicに関連付けることができますが、同じTopicを同時に複数の同期タスクで使用することはできません。そうしないと、データの不整合が発生し、消費されるデータが複数の異なるタスクのデータとなり、タスクの再試行などのシナリオでデータ処理の異常やタスクエラーが発生する可能性があります。
デフォルトではロックあり同期方式を採用し、フルデータエクスポート段階ではソースデータベースにグローバルロック（FTWRL）をかけ、テーブルロック中は短時間（秒単位）書き込み操作をブロックします。
ユーザーが同期中に特定のテーブルに対してrename操作を行うことが確実な場合、データはターゲットCkafkaに配信され、パーティションルールは新しいテーブル名に基づいてtopicとpartitionをマッチングします。
全量同期段階では、10万件のデータを同期するごとに、ターゲット側Kafkaにcheckpointメッセージが1件挿入され、現在のデータ同期位置を示します。
同期タスクの前に、ソースデータベースに既に存在する全量データは、データ書き込みの正確な時間がないため、コンシューマー側DemoではhappenAtフィールドに1970-01-01 08:00:00と表示されますが、この時間フィールドは気にする必要はありません。増分データを消費する際、時間フィールドは正しく表示されます。
DTSは全量データ同期段階で、ソースデータベースのデータをエクスポートし、ターゲット側Ckafkaにインポートします。いずれもutf8mb4文字セットを使用し、文字セットが認識されないことによる文字化けを防ぎます。
ターゲットTopicは空にすることをお勧めします。また、同期タスク中はターゲットTopicに書き込みを行わないでください。そうしないと、メッセージの不整合やタスクエラーが発生する可能性があります。
DTSがターゲット側Kafkaに同期する単一メッセージには性能上限があります。ユーザーはソースデータベース内の単一行データを8MB以下に抑えることをお勧めします。そうしないと、同期中にエラーが発生する可能性があります。
データ同期時、DTSは同期タスクを実行するアカウントを使用してソースデータベースにシステムデータベース
__tencentdb__を書き込み、トランザクションマークIDなどのメタ情報を記録します。ソースデータベースにおける__tencentdb__への読み書き権限を確保する必要があります。後続のデータ比較問題を特定できるようにするため、同期タスク終了後もソースデータベース内の
__tencentdb__は削除されません。__tencentdb__システムデータベースが占有する容量は非常に小さく、ソースデータベースのストレージ容量の約1000分の1から1万分の1です（例えば、ソースデータベースが50GBの場合、__tencentdb__システムデータベースは約5MB～50MBです）。また、シングルスレッドと接続待機メカニズムを採用しているため、ソースデータベースのパフォーマンスにはほとんど影響を与えず、リソースを奪い合うこともありません。前提条件
ソースデータベースとターゲットデータベースが同期機能とバージョン要件を満たしていることを確認してください。確認にはデータ同期がサポートするデータベースを参照してください。
同期タスクのアカウントに必要なソースデータベースの権限は以下の通りです：
GRANT RELOAD,LOCK TABLES,REPLICATION CLIENT,REPLICATION SLAVE,SHOW VIEW,PROCESS,SELECT ON *.* TO 'アカウント'@'%' IDENTIFIED BY 'パスワード';GRANT ALL PRIVILEGES ON `__tencentdb__`.* TO 'アカウント'@'%';FLUSH PRIVILEGES;
ターゲットCkafkaでメッセージ保持期間とメッセージサイズの上限を変更します。
メッセージ保存期間は3日に設定することをお勧めします。保存期間を過ぎたデータは削除されます。ユーザーは設定した期間内に速やかに消費してください。メッセージサイズの上限、つまりCkafkaが受信できる単一メッセージのメモリサイズは、ソースデータベースのテーブル内の単一行データの最大値より大きく設定する必要があります。これにより、ソースデータベースのデータがすべて正常にCKafkaに配信されます。
アプリケーション制限
基本テーブル、ビュー、ストアドプロシージャ、および関数の同期のみをサポートします。
ソース側が非GTIDインスタンスの場合、DTSはソース側のHA切り替えをサポートしません。ソース側のMySQLで切り替えが発生すると、DTSの増分同期が中断される可能性があります。
InnoDB、MyISAM、TokuDBの3種類のデータベースエンジンのみ同期をサポートします。これら以外のデータエンジンテーブルが存在する場合、デフォルトでスキップされ同期されません。
相互に関連するデータオブジェクトは一緒に同期する必要があります。そうしないと同期が失敗します。例えば、主キー・外部キー関連テーブルなどがあります。
1つのトランザクションにDML文とDDL文が同時に含まれるシナリオはサポートされていません。この状況が発生した場合、タスクはエラーを報告します。
Geometry関連のデータ型はサポートされていません。このタイプのデータが検出された場合、タスクはエラーを報告します。
ALTER VIEWステートメントはサポートされていません。このステートメントが検出された場合、タスクはスキップされ、同期されません。データの一貫性を保証するため、Kafkaへの同期はデフォルトでロック付き同期を採用します。ただし、ソースデータベースがAlibaba Cloud MySQL5.5/5.6、AWS MySQL 5.7.18以前のバージョンの場合、ロックを追加できず、ターゲットCkafkaに同期される一部のデータが重複する可能性があります。重複データが気になる場合は、データを消費する際に重複排除ロジックを設定してください。
同期タスク中にタスクの再起動が発生した場合、ターゲット側Ckafkaに同期されたデータに重複が生じる可能性があります。
DTSは最小データユニット単位で同期を行います（全量段階では単一テーブルオブジェクトの1ブロックのデータが最小データユニットとなり、増分段階では1つのポジションがマークされるごとに1つのデータユニットとなります）。再起動時に、ちょうど1つのデータユニットの同期が完了している場合は、データの重複は発生しません。再起動時に、1つのデータユニットがまだ同期中である場合は、再度起動した後にこのデータユニットを再同期してデータの整合性を保つ必要があるため、データの重複が発生します。
同期オブジェクトの変更、タスクエラー時に操作列の再試行ボタンを手動でクリックする操作、設定変更、タスクの一時停止後の再開、隔離後の隔離解除、これらのシナリオではいずれもタスクの再起動が発生します。
ユーザーが重複データを気にする場合は、データを消費する際に重複排除ロジックを設定してください。
ターゲット側Kafkaのバージョンは、現在Tencent Cloudで購入可能なプロフェッショナル版とスタンダード版のみサポートされており、バージョン1.1.1以降が対象です。すでに販売終了となった物理マシンのスタンダード版はターゲット側としてサポートされていません。
操作制限
同期中に以下の操作を行わないでください。行った場合、同期タスクが失敗する可能性があります。
ソースデータベースおよびターゲットCkafka内のユーザー情報（ユーザー名、パスワード、権限を含む）とポート番号を変更、削除しないでください。
ソースデータベースにBinlog形式が
STATEMENTのデータを書き込まないでください。ソースデータベース上でBinlogを削除する操作を実行しないでください。
ターゲット側で同期対象として選択済みのTopicにデータを書き込まないでください。
同期サポート対象のSQL操作
操作タイプ | SQL操作文 |
DML | INSERT、UPDATE、DELETE |
DDL | CREATE DATABASE、DROP DATABASE、ALTER DATABASE、CREATE TABLE、ALTER TABLE、DROP TABLE、TRUNCATE TABLE、CREATE VIEW、DROP VIEW、CREATE INDEX、DROP INDEX |
説明：
CREATE TABLE テーブル名 AS SELECT文は現在サポートされていません。
環境要件
タイプ | 環境要件 |
ソースデータベース要件 | ソースデータベースとターゲットデータベースのネットワークが接続できる必要があります。 インスタンスパラメータの要件： ソースデータベースのserver_idパラメータは手動で設定する必要があります。また、値を0に設定することはできません。 ソースデータベースのテーブルのrow_formatをFIXEDに設定することはできません。 ソースデータベース変数connect_timeoutの設定値は10より大きい必要があります。 Binlogパラメータの要件： ソース側のlog_bin変数はONに設定する必要があります。 ソース側のbinlog_format変数はROWに設定する必要があります。 ソース側のbinlog_row_image変数はFULLに設定する必要があります。 MySQL 5.6以降のバージョンでは、gtid_mode変数がONでない場合に警告が表示されます。gtid_modeを有効にすることをお勧めします。 do_db、ignore_dbを設定することはできません。 ソースインスタンスがレプリカの場合、log_slave_updates変数をONに設定する必要があります。 ソースデータベースのBinlogログは少なくとも3日以上保持することをお勧めします。そうしないと、タスクの一時停止/中断時間がBinlogログの保持時間を超えた場合、タスクの再開ができなくなり、タスクが失敗する可能性があります。 外部キー依存関係： 外部キー依存関係はNO ACTION、RESTRICTの2種類に設定することをお勧めします。他のタイプはデータの一貫性の結果に影響を与える可能性があります。 一部のデータベース・テーブルを同期する場合、外部キー依存関係のあるテーブルはすべて揃っている必要があります。 環境変数innodb_stats_on_metadataはOFFに設定する必要があります。 |
ターゲットCkafka要件 | ターゲットCkafkaは正常に接続できます。 ターゲットCkafkaで設定するメッセージサイズの上限は、ソースデータベースのテーブル内の単一行データの最大値より大きくする必要があります。 |
操作手順
MySQL/MariaDB/Percona/TDSQL-C MySQLからTencent CloudデータベースCkafkaへの同期は、各シナリオの同期要件と操作手順がほぼ同じであるため、本章ではMySQLからCkafkaへのデータ同期を例として説明します。その他のシナリオについては、関連内容を参照してください。
1. データ同期購入ページにログインし、対応する設定を選択して、今すぐ購入をクリックします。
パラメータ | 説明 |
課金モデル | 年額・月額と従量課金をサポートします。 |
ソースインスタンスタイプ | MySQLを選択します。購入後は変更できません。 |
ソースインスタンスの地域 | ソースインスタンスが所在する地域を選択してください。購入後は変更できません。ソースデータベースが自己構築データベースの場合、自己構築地域に最も近い地域を選択してください。 |
ターゲットインスタンスタイプ | Kafkaを選択します。購入後は変更できません。 |
ターゲットインスタンスの地域 | 目的インスタンスの所在地域を選択します。購入後は変更できません。 |
仕様 |
2. 購入完了後、データ同期リストに戻ると、作成したばかりのデータ同期タスクが表示されます。操作列の設定をクリックして、同期タスクの設定ページに進みます。
3. 同期タスクの設定ページで、ソースインスタンス、アカウントパスワードを設定し、ターゲットインスタンス、アカウントとパスワードを設定し、接続性をテストした後、次へをクリックします。
設定項目 | パラメータ | 説明 |
タスク設定 | タスク名 | DTSはタスク名を自動生成します。ユーザーは実際の状況に応じて設定できます。 |
| 実行モード | 即時実行と定時実行の2つのモードをサポートします。 |
ソースインスタンスの設定 | ソースインスタンスタイプ | 購入時に選択したソースインスタンスタイプは変更できません。 |
| ソースインスタンスの地域 | 購入時に選択したソースインスタンスの所在地域は変更できません。 |
| アクセスタイプ | 公衆網：ソースデータベースにはパブリックIPアドレスでアクセスできます。 CVM自社構築：ソースデータベースはTencent CVM上に配置されています。 Direct Connect：ソースデータベースはDirect Connectを介してTencent Cloud VPCと接続できます。 VPNアクセス：ソースデータベースはVPN Connectionsを介してTencent Cloud VPCと接続できます。 クラウドデータベース：ソースデータベースはTencent Cloudデータベースインスタンスに属します。 CCN：ソースデータベースはCCNを介してTencent Cloud VPCと接続できます。 |
| パブリックネットワーク | ホストアドレス：ソースデータベースのIPアドレスまたはドメイン名。 ポート：ソースデータベースが使用するポート。 |
| CVMでの自己構築 | CVMインスタンス：CVMのインスタンスID。 ポート：ソースデータベースが使用するポート。 |
| Direct Connect (DC) | VPC Direct Connectゲートウェイ/：Direct Connectアクセス時はVPC Direct Connectゲートウェイのみサポートされます。ゲートウェイに関連付けられたネットワークタイプを確認してください。 VPC：VPC Direct ConnectゲートウェイおよびVPNゲートウェイに関連付けられたVPCとサブネットを選択します。 ホストアドレス：ソースデータベースのIPアドレス。 ポート：ソースデータベースが使用するポート。 |
| VPNアクセス | VPNゲートウェイ：VPNゲートウェイ。VPNゲートウェイを介してアクセスするVPNゲートウェイインスタンスを選択してください。 VPC：VPC Direct ConnectゲートウェイおよびVPNゲートウェイに関連付けられたVPCとサブネットを選択します。 ホストアドレス：ソースデータベースのIPアドレス。 ポート：ソースデータベースが使用するポート。 |
| クラウドデータベース | クラウドデータベースインスタンス：ソースデータベースのインスタンスID。 |
| CCN | ホストアドレス：ソースデータベースのホストIPアドレス。 ポート：ソースデータベースが使用するポート。 VPC CCN：CCNインスタンス名。 アクセスVPC：アクセスVPCとは、CCN内でサブスクリプションリンクに接続されるVPCを指します。CCNに関連付けられているすべてのVPCの中から、ソースデータベースが属するVPC以外のVPCを選択してください。
例えば、広州地域のデータベースがソースデータベースの場合、アクセスVPCには成都VPCや上海VPCなど、他の地域を選択します。 サブネット：選択済みのVPCネットワークのサブネット名。 アクセスVPC地域：タスク購入時に選択したソースデータベースの地域とアクセスVPCの地域は一致している必要があります。一致していない場合、DTSはタスク購入時に選択したソースデータベースの地域をアクセスVPCの地域に変更します。 |
| アカウント/パスワード | アカウント/パスワード：ソースデータベースのアカウント、パスワード。 |
ターゲットインスタンスの設定 | ターゲットインスタンスタイプ | 購入時に選択したターゲットインスタンスタイプは変更できません。 |
| ターゲットインスタンスの地域 | 購入時に選択したターゲットインスタンスの地域は変更できません。 |
| アクセスタイプ | お客様のシナリオに応じて選択してください。このシナリオでは「Ckafkaインスタンス」を選択します。 |
| インスタンスID | ターゲットインスタンスIDを選択します。 |
4. 同期オプションと同期オブジェクトの設定ページで、データ初期化オプション、データ同期からKafkaへの戦略、データ同期オプション、同期オブジェクトオプションを設定します。設定が完了したら、保存して次へをクリックします。
カスタムTopic名
単一Topicへの集中配信
設定項目 | パラメータ | 説明 |
データ初期化オプション | 初期化タイプ | 構造初期化：同期タスクの実行時に、まずソースインスタンスのテーブル構造をターゲットインスタンスに初期化します。 全量データ初期化：同期タスクの実行時に、まずソースインスタンスのデータをターゲットインスタンスに初期化します。全量データ初期化のみを選択した場合、ユーザーは事前にターゲットデータベースにテーブル構造を作成しておく必要があります。 デフォルトでは両方ともチェックされており、実際の状況に応じてキャンセルできます。 |
| Kafka に配信するデータ形式 | Avroはバイナリ形式で、消費効率がより高くなっています。JSONは軽量のテキスト形式で、よりシンプルで使いやすくなっています。 |
データ同期からKafkaへのポリシー | Topic同期ポリシー | カスタムTopic名：配信するTopic名を設定します。設定後、ターゲット側のKafkaがそのTopicを自動的に作成します。実現効果として、同期データをそのTopic配下の異なるパーティションにランダムに配信します。ターゲット側のKafkaでTopicの作成に失敗した場合、タスクはエラーを報告します。 単一Topicへの集中配信：ターゲット側に既存のTopicを選択し、複数のパーティションポリシーに従って配信します。単一パーティション、テーブル名によるパーティショニング、テーブル名+PKによるパーティショニングをサポートし、異なるパーティションに配信します。 |
| カスタムTopic名ルール | ユーザーが複数のルールを追加すると、カスタムTopicマッチルールは上から順番にマッチングします。設定されたデータベース名とテーブル名のルールがすべて一致した場合のみ、そのルールに対応するTopicに発送されます。データベース・テーブルのマッチルールに一致しない場合、最後のルールのTopicに発送されます。複数のデータベース・テーブルのマッチルールに一致した場合、すべてのマッチルールのTopicに発送されます。 説明： マッチルールはデータベース・テーブル名の大文字と小文字を区別します。 ソースデータベースでlower_case_table_names=0に設定した場合、データベース・テーブルのマッチルール内のデータベース・テーブル名は、ソースデータベース内の名前と大文字・小文字を厳密に一致させる必要があります。 ソースデータベースでlower_case_table_names=1に設定した場合、データベース・テーブル名は小文字に統一して変換され、データベース・テーブルのマッチルール内のデータベース・テーブル名は小文字で統一して入力されます。 例1：データベースインスタンスXにおいて、UsersデータベースにStudent、Teacherテーブルがあり、UsersデータベースのデータをすべてTopic_Aに配信する必要があります。設定ルールは以下の通りです： Topic名に Topic_Aを入力し、データベース名のマッチングモードに^Users$を入力し、テーブル名のマッチングモードに .*を入力します。Topic名に Topic_defaultを入力し、データベース名のマッチングモード（マッチルールに一致しない残りのデータベース）、テーブル名のマッチングモード（マッチルールに一致しない残りのテーブル）を入力します。例2：データベースインスタンスXにおいて、UsersデータベースにStudent、Teacherテーブルがあり、StudentテーブルのデータをTopic_Aに、TeacherテーブルのデータをTopic_defaultに配信する必要があります。設定ルールは以下の通りです： Topic名に Topic_Aを入力し、データベース名のマッチングモードに^Users$を入力し、テーブル名のマッチングモードに^Student$を入力します。Topic名に Topic_defaultを入力し、データベース名のマッチングモード（マッチルールに一致しない残りのデータベース）、テーブル名のマッチングモード（マッチルールに一致しない残りのテーブル）を入力します。 |
| 単一Topicへの集中配信ルール | 指定されたTopicを選択すると、システムは以下の指定ポリシーに従ってパーティショニングを行います。 すべてPartition0に配信します：ソースデータベースの同期データをすべて最初のパーティションに配信します。 テーブル名によるパーティショニング：ソースデータベースの同期データをテーブル名に従ってパーティショニングします。設定後、同じテーブル名のデータは同じパーティションに書き込まれます。 テーブル名+PKによるパーティショニング：ソースデータベースの同期データをテーブル名+PKに従ってパーティショニングします。ホットデータに適しており、設定後、ホットデータのテーブルは、テーブル名+PKの方式でデータを異なるパーティションに分散し、並行消費効率を向上させます。 |
| DDLを格納するTopic | 任意です。ユーザーがソースデータベースのDDL操作を指定したTopicに個別に発送する必要がある場合、ここで選択して設定できます。
設定後、デフォルトで選択済みTopicのPartition0に発送されます。設定されていない場合は、上で選択したTopicルールに従って発送されます。 |
データ同期オプション
設定項目 | パラメータ | 説明 |
データ同期オプション | 同期操作タイプ | サポートされる操作：Insert、Update、Delete、DDL。「DDLカスタム」をチェックすると、必要に応じて異なるDDL同期ポリシーを選択できます。詳細については、SQLフィルタリングポリシーの設定を参照してください。 |
同期オブジェクトオプション | ソースインスタンスのテーブルオブジェクト | 同期対象のオブジェクトを選択します。基本データベース・テーブル、ビュー、ストアドプロシージャ、関数をサポートします。高度なオブジェクトの同期は1回限りのアクションであり、タスク開始前にソースデータベースに既に存在する高度なオブジェクトのみを同期できます。タスク開始後に新しく追加された高度なオブジェクトはターゲットデータベースに同期されません。詳細については、高度なオブジェクトの同期を参照してください。 |
| 選択済みオブジェクト | 左側で同期オブジェクトを選択した後、  |
| Online DDL一時テーブルを同期するかどうか | gh-ost、pt-oscツールを使用してソースデータベース内のテーブルに対してOnline DDL操作を実行する場合、DTSはOnline DDLの変更によって生成された一時テーブルをターゲットデータベースに移行することをサポートします。 gh-ostをチェックすると、DTSはgh-ostツールが生成した一時テーブル名（`_表名_ghc`、`_表名_gho`、`_表名_del`）をターゲットデータベースに移行します。 pt-oscをチェックすると、DTSはpt-oscツールが生成した一時テーブル名（`_表名_new`、`_表名_old`）をターゲットデータベースに移行します。 |
5. 検証タスクページで、検証を完了し、すべての検証項目に合格した後、タスクを開始をクリックします。
検証タスクが合格しない場合は、検証不合格の対処方法を参照して問題を修正した後、検証タスクを再実行します。
失敗：検証項目のチェックが不合格であり、タスクがブロックされていることを示します。問題を修正した後、検証タスクを再実行する必要があります。
警告：検証項目のチェックが要件を完全には満たしていないことを示します。タスクを続行できますが、業務に一定の影響があります。ユーザーはプロンプトに基づいて、警告項目を無視するか、問題を修正してから続行するかを自身で評価する必要があります。
6. データ同期タスクリストに戻り、タスクは実行中状態に入ります。
説明
操作列の詳細 > 終了を選択すると、同期タスクを終了できます。データ同期が完了したことを確認してからタスクを終了してください。
7. （オプション）タスク名をクリックすると、タスク詳細ページに入り、タスクの初期化状態と監視データを確認できます。
よくあるご質問
増分同期中、ソースデータベースの同期オブジェクトに長時間書き込みがない場合、監視に大きな遅延が表示されます。
ユーザーはまず、ソースデータベースで同期対象として選択したオブジェクトに長期間データが書き込まれていないかどうかを確認してください（他の未同期のデータベース・テーブルオブジェクトにはデータが書き込まれています）。書き込まれていないことを確認した場合は、この問題を無視してください。このようなシナリオでは監視指標が歪んでいます。以下の説明を参照してください。書き込みがある場合は、チケットを送信して処理してください。
MySQLシリーズデータベースからKafkaへの同期リンクでは、DTSはまずソースデータベースのBinlogを取得し、解析後に対象オブジェクトの関連データを識別して、ターゲットKafkaに同期する必要があります。ソースデータベースで同期対象として選択したデータベース・テーブルオブジェクトに長期間データが書き込まれておらず、他の未同期のデータベース・テーブルオブジェクトにはデータが書き込まれている場合、DTSがBinlogを取得して解析した結果、ターゲット側に同期する必要がないため、ターゲット側のデータが更新されず、遅延が表示されます。ユーザーは、ソースデータベースで同期対象として選択したオブジェクトのデータを更新して指標を更新し、実際の指標データを取得することができます。
後続操作
データをターゲット側Kafkaに同期した後、データを消費できます。データ消費のプロセスを迅速にテストし、データ形式の解析方法を理解できるように、消費サンプルDemoを提供しています。詳細については、データ消費操作ガイドを参照してください。
フィードバック