TiDB から AnalyticDB for MySQL 3.0 へのデータ同期 - Data Transmission Service

Data Transmission Service (DTS) を使用して、自己管理 TiDB データベースから AnalyticDB for MySQL V3.0 クラスターにデータを同期し、リアルタイム分析を実現します。

仕組み

DTS は、変更データキャプチャ (CDC) バッファーとして機能する Kafka クラスターを介して TiDB から増分変更を読み取り、そのデータを AnalyticDB for MySQL V3.0 に書き込みます。完全なパイプラインは次のとおりです。

TiDB → Kafka クラスター → DTS → AnalyticDB for MySQL V3.0

TiDB はバイナリログを DTS に直接公開しません。次のいずれかの CDC コンポーネントを設定して、増分変更を Kafka トピックに公開する必要があります。

TiDB Binlog：Pump および Drainer コンポーネントを使用します
TiCDC：Canal-JSON フォーマットを使用します

DTS は Kafka トピックのパーティション 0 から読み取り、変更をターゲットクラスターに適用します。

制限事項

タスクを設定する前に、これらの制限事項を確認してください。

ソースデータベースの制限

制限	詳細
プライマリキーまたは一意制約が必須	同期するテーブルには、すべてのフィールドが一意である PRIMARY KEY または UNIQUE 制約が必要です。この制約がないテーブルでは、ターゲットでレコードが重複する可能性があります。
同期中の DDL 制限	スキーマ同期または完全データ同期中に、データベースまたはテーブルスキーマを変更する DDL 文を実行しないでください。タスクは失敗します。
プレフィックスインデックス長の損失	TiDB はメタデータにプレフィックスインデックス長を保存しません。データが AnalyticDB for MySQL V3.0 に同期されると、この長さは失われ、ターゲットインスタンスが失敗する可能性があります。プレフィックスインデックスを使用するテーブルについては、手動でプレフィックスインデックス長を修正してください。
タスクあたりのテーブル制限	個々のテーブル (データベース全体ではない) を選択して名前を変更する場合、1 つのタスクでサポートされるテーブルは最大 1,000 です。この制限を超える場合は、複数のタスクを設定するか、代わりにデータベース全体を同期してください。
Kafka のセットアップが必須	DTS タスクを作成する前に、Kafka クラスターをデプロイし、TiDB Binlog または TiCDC をインストールしてください。
アウトバウンド帯域幅	TiDB をホストするサーバーには、十分なアウトバウンド帯域幅が必要です。帯域幅が不足すると、同期速度が低下します。

宛先とタスクの制限

制限	詳細
Kafka パーティション	DTS は Kafka トピックのパーティション 0 からのみ読み取ります。トピックは、パーティションを 1 つだけにして作成してください。
カスタムプライマリキーが必須	AnalyticDB for MySQL V3.0 でカスタムプライマリキーを指定するか、[データベース、テーブル、および列の設定] ステップで [プライマリキー列] を設定します。そうしないと、タスクが失敗する可能性があります。
ディスク使用率のしきい値	AnalyticDB for MySQL V3.0 クラスターのいずれかのノードでディスク使用率が 80% を超えると、DTS タスクは遅延し、エラーが返されます。開始前に必要なディスク領域を見積もってください。
バックアップの競合	DTS の実行中にターゲットクラスターがバックアップされている場合、タスクは失敗します。
異種データ型	TiDB と AnalyticDB for MySQL V3.0 のデータ型は 1 対 1 でマッピングされません。詳細については、「初期スキーマ同期のデータ型マッピング」をご参照ください。
FLOAT/DOUBLE の精度	DTS は `ROUND(COLUMN,PRECISION)` を使用して FLOAT および DOUBLE 値を取得します。FLOAT のデフォルトの精度は 38 桁、DOUBLE は 308 桁です。これらのデフォルト値が要件を満たしていることを確認してください。
マテリアライズドビューはサポート対象外	スキーマ同期中、DTS はマテリアライズドビューをターゲットの AnalyticDB for MySQL インスタンスに同期することをサポートしていません。マテリアライズドビューを使用するには、同期が完了した後にターゲットインスタンスで手動で作成してください。
タスクオフセットの初期化	タスクを作成した後、速やかにソースデータベースで操作を実行するか、テストデータを挿入してください。これにより、タスクオフセットが更新されます。このステップをスキップすると、過度のレイテンシーによりタスクが失敗する可能性があります。
他のソースからのデータ	同期中に他のソースからターゲットクラスターにデータを書き込まないでください。データの不整合が発生します。
失敗したタスクの再開	DTS は失敗したタスクを最大 7 日間自動的に再試行します。ワークロードをターゲットクラスターに切り替える前に、失敗したタスクを停止またはリリースしてください。または、`REVOKE` を実行して、ターゲットデータベースから DTS の書き込み権限を削除し、再開されたタスクが新しいデータを上書きできないようにします。
完全データ同期のパフォーマンス	完全データ同期は、ソースとターゲットの両方のデータベースの負荷を増加させます。CPU 使用率が 30% 未満のオフピーク時に実行してください。
完全同期後の表領域サイズ	完全データ同期中の同時 INSERT 操作は、テーブルの断片化を引き起こします。ターゲットの表領域はソースよりも大きくなります。
DDL の失敗	ターゲットデータベースで DDL 文が失敗した場合でも、タスクは続行されます。失敗した DDL 文を表示するには、タスクログを確認してください。
DTS テクニカルサポートの SLA	DTS タスクの実行に失敗した場合、DTS テクニカルサポートは 8 時間以内にタスクの復元を試みます。復元中、タスクが再起動されたり、タスクパラメーターが変更されたりする場合があります。変更される可能性があるのはタスクパラメーターのみで、データベースパラメーターは変更されません。

説明

送信先が AnalyticDB for MySQL クラスターの場合、DTS は、AnalyticDB for MySQL でネイティブにサポートされているデータの型の書き込みのみをサポートします。これには、基本データ型や、ARRAY、MAP、JSON などの複合データ型が含まれます。 MULTIVALUE などのデータの型はサポートされていません。

増分同期でサポートされる SQL 操作

操作タイプ	サポートされる文
DML	INSERT、UPDATE、DELETE
DDL	CREATE TABLE、DROP TABLE、RENAME TABLE、TRUNCATE TABLE、ADD COLUMN、DROP COLUMN

DTS が AnalyticDB for MySQL V3.0 に UPDATE 文を書き込む際、自動的に REPLACE INTO に変換します。UPDATE 文がプライマリキーに影響する場合、DTS はそれを DELETE とそれに続く INSERT に変換します。

課金

同期タイプ	コスト
スキーマ同期と完全データ同期	無料
増分データ同期	課金概要有料です。詳細については、「」をご参照ください。

前提条件

開始する前に、以下を確認してください。

TiDB データベースの合計データサイズよりも大きい利用可能なストレージを持つ AnalyticDB for MySQL V3.0 クラスター。詳細については、「クラスターの作成」をご参照ください
TiDB サーバーからアクセス可能な Kafka クラスターまたは ApsaraMQ for Kafka インスタンスがデプロイされていること
必要なデータベースアカウントの権限 (詳細については、「必要な権限」をご参照ください)

必要な権限

データベース	必要な権限	リファレンス
TiDB	同期するオブジェクトに対する SELECT、SHOW VIEW	権限管理
AnalyticDB for MySQL V3.0	ターゲットデータベースに対する読み取りおよび書き込み権限	データベースアカウントの作成

増分データキャプチャの設定

DTS では、TiDB からの増分変更イベントを Kafka トピックに公開する必要があります。次のいずれかの方法を選択してください。

方法の選択

	TiDB Binlog	TiCDC
コンポーネント	Pump + Drainer	TiCDC (TiUP 経由で管理)
出力フォーマット	デフォルトの TiDB バイナリログフォーマット	Canal-JSON フォーマット
最適なケース	既存の TiDB Binlog デプロイメント	新規セットアップおよび TiDB v4.0 以降
DTS 設定	TiDB データベースのデフォルトのバイナリログフォーマットを使用	TiCDC Canal-JSON フォーマットを使用

Kafka クラスターの準備

どちらの方法でも Kafka クラスターが必要です。次のいずれかを使用してください。

自己管理 Kafka クラスター：Apache Kafka をデプロイします。詳細については、「Apache Kafka ドキュメント」をご参照ください
ApsaraMQ for Kafka インスタンス：インスタンスを作成します。詳細については、「クイックスタート概要」をご参照ください

どのオプションを選択するかにかかわらず：

ネットワーク遅延を最小限に抑えるために、TiDB サーバーと同じネットワークまたは Virtual Private Cloud (VPC) に Kafka クラスターをデプロイします。
Kafka ブローカーのパラメーター message.max.bytes と replica.fetch.max.bytes を大きな値に設定し、コンシューマーのパラメーター fetch.message.max.bytes を一致する大きな値に設定します。これにより、クラスターが TiDB からのバイナリログデータの量を処理できるようになります。詳細については、「Kafka 設定リファレンス」をご参照ください。

クラスターを設定した後、パーティションが 1 つだけのトピックを作成します。DTS はパーティション 0 からのみ読み取るため、複数のパーティションがあるとデータ損失が発生します。

オプション 1：TiDB Binlog

TiDB と同じ内部ネットワーク内のサーバーに Pump と Drainer をデプロイします。詳細については、「TiDB Binlog クラスターのデプロイ」をご参照ください。
Drainer 設定ファイルを編集して、ダウンストリームシンクとして Kafka クラスターを指すようにします。詳細については、「Binlog Consumer Client ユーザーガイド」をご参照ください。続行する前に、TiDB サーバーが Kafka クラスターに接続できることを確認してください。
DTS サーバーの CIDR ブロックを TiDB データベースの許可リストに追加します。「DTS サーバーの CIDR ブロックを追加する」をご参照ください。

オプション 2：TiCDC

TiUP を使用して TiCDC をインストールします。TiDB クラスターに新しい TiCDC ノードを追加するか、既存のノードをスケールアウトします。詳細については、「TiCDC のデプロイとメンテナンス」をご参照ください。
TiDB から Kafka クラスターに増分データをレプリケートするための changefeed を作成します。tiup cdc cli changefeed create を使用し、Canal-JSON sink URI を指定します。詳細については、「Kafka へのデータレプリケーションKafka へのデータ同期」をご参照ください。続行する前に、TiDB サーバーが Kafka クラスターに接続できることを確認してください。

データ同期タスクの作成

ステップ 1：データ同期ページへの移動

DTS コンソールまたは DMS コンソールを使用します。

DTS コンソール

DTS コンソールにログインします。DTS コンソール
左側のナビゲーションウィンドウで、[データ同期] をクリックします。
左上隅で、同期インスタンスが存在するリージョンを選択します。

DMS コンソール

DMS コンソールのレイアウトによって手順が異なる場合があります。「シンプルモード」および「DMS コンソールのレイアウトとスタイルをカスタマイズする」をご参照ください。

DMS コンソールにログインします。DMS コンソール
トップナビゲーションバーで、ポインターを[Data + AI]に合わせ、 [DTS (DTS)] > [データ同期]を選択します。
[データ同期タスク] の右側のドロップダウンリストから、同期インスタンスを配置するリージョンを選択します。

ステップ 2：ソースデータベースとターゲットデータベースの設定

[タスクの作成] をクリックし、次のパラメーターを設定します。

ソースデータベース

パラメーター	説明
タスク名	DTS タスクの名前。DTS が自動的に生成します。簡単に識別できるように、わかりやすい名前を指定してください。名前は一意である必要はありません。
DMS データベースインスタンスの選択	既存の登録済みデータベースを選択するか、この項目を空白のままにして以下のパラメーターを入力します。データベースを登録するには、DMS コンソールで [DMS データベースインスタンスの追加] をクリックします。DTS コンソールでは、[データベース接続] ページを使用します。「Alibaba Cloud データベースインスタンスを登録する」および「サードパーティのクラウドサービスでホストされているデータベースまたは自己管理データベースを登録する」をご参照ください。
データベースタイプ	[TiDB] を選択します。
アクセス方法	TiDB がデプロイされている場所と一致するアクセス方法を選択します。この例では、[ECS 上の自己管理データベース] を使用します。他のアクセス方法については、まず必要な環境を準備してください。詳細については、「準備の概要」をご参照ください。
インスタンスリージョン	TiDB が存在するリージョン。
ECS インスタンス ID	TiDB をホストしている ECS インスタンスの ID。
ポート番号	TiDB サービスポート。デフォルト: 4000。
データベースアカウント	必要な権限を持つ TiDB アカウント。
データベースパスワード	TiDB アカウントのパスワード。
増分データの移行	[はい]アラート通知設定。これは変更できません。増分データなしで同期するには、代わりにデータ移行タスクを作成します。詳細については、「自己管理 TiDB データベースから AnalyticDB for MySQL V3.0 クラスターへのデータ移行」をご参照ください。
Kafka クラスタータイプ	Kafka クラスターのデプロイ場所に応じたアクセス方法を選択します (この例では [ECS 上の自己管理型データベース] を使用)。 [Express Connect、VPN ゲートウェイ、または Smart Access Gateway] を選択した場合は、[接続済み VPC] から VPC を選択し、[ドメイン名または IP] も指定します。
Kafka データソースコンポーネント	TiDB データベースのデフォルト binlog フォーマット「」で設定した内容に基づいて、[TiDB データベースのデフォルトのバイナリログ形式を使用]増分データキャプチャの設定 (TiDB Binlog の場合) または (TiCDC の場合) を選択します。
ECS インスタンス ID	Kafka クラスターがデプロイされている ECS インスタンスの ID。
ポート番号	Kafka サービスポート。
Kafka クラスターアカウント / Kafka クラスターパスワード	Kafka のユーザー名とパスワード。認証が有効になっていない場合は空白のままにします。
Kafka バージョン	Kafka のバージョンです。バージョンが 1.0 以降の場合は、[1.0] を選択してください。
暗号化	セキュリティ要件に応じて、[非暗号化] または SCRAM-SHA-256 を選択します。
トピック	TiDB からの増分データを受信する Kafka トピック。

ターゲットデータベース

パラメーター	説明
DMS データベースインスタンスの選択	既存の登録済みデータベースを選択するか、このフィールドを空白にして以下のパラメーターを入力します。
データベースタイプ	AnalyticDB for MySQL 3.0 を選択します。
アクセス方法	Alibaba Cloud インスタンスを選択します。
インスタンスリージョン	AnalyticDB for MySQL V3.0 クラスターが存在するリージョン。
インスタンス ID	ターゲットの AnalyticDB for MySQL V3.0 クラスターの ID。
データベースアカウント	読み取りおよび書き込み権限を持つデータベースアカウント。
データベースパスワード	データベースアカウントのパスワード。

ステップ 3：接続テスト

[接続をテストして続行] をクリックします。[DTS サーバーの CIDR ブロック] ダイアログボックスで、[接続テスト] をクリックします。

DTS の CIDR ブロックがデータベースのセキュリティ設定にまだ追加されていない場合は、続行する前に追加してください。詳細については、「DTS サーバーの CIDR ブロックを追加する」をご参照ください。

ステップ 4：同期オブジェクトの設定

[オブジェクトの設定] ステップでは、以下のパラメーターを設定します。

パラメーター	説明
同期タイプ	[スキーマ同期]、[完全データ同期]、[増分データ同期] を選択します。3 つすべてが必須です。[増分データ同期] はデフォルトで選択されていますが、他の 2 つも選択する必要があります。完全データ同期は、増分同期のための既存データのベースラインを提供します。
競合するテーブルの処理モード	[事前チェックとエラー報告]オブジェクト名のマッピング (デフォルト)：宛先に同じ名前のテーブルが存在する場合、事前チェックは失敗します。オブジェクト名マッピングを使用して、宛先テーブルを削除せずに競合を解決します。詳細については、「」をご参照ください。[エラーを無視して続行]：同じテーブル名に対する事前チェックをスキップします。完全同期中、宛先の既存レコードは保持されます。増分同期中、宛先レコードは上書きされます。データの不整合を引き起こす可能性があるため、注意して使用してください。
宛先インスタンスのオブジェクト名の大文字/小文字	宛先のデータベース、テーブル、列名の大文字/小文字をコントロールします。デフォルト：[DTS のデフォルトポリシー]宛先インスタンスのオブジェクト名の大文字/小文字の指定。詳細については、「」をご参照ください。
ソースオブジェクト	同期するデータベースまたはテーブルを選択し、矢印アイコンをクリックして [選択したオブジェクト] に移動します。
選択したオブジェクト	宛先でオブジェクトの名前を変更するには、オブジェクトを右クリックしてオブジェクト名マッピングを使用します。一度に複数のオブジェクトの名前を変更するには、[バッチ編集] をクリックします。SQL の WHERE 条件で行をフィルターするには、テーブルを右クリックして条件を指定します。詳細については、「フィルター条件の指定」をご参照ください。オブジェクトの名前を変更すると、依存オブジェクトが中断される可能性があります。

ステップ 5：詳細設定

[次へ: 詳細設定] をクリックし、以下を設定します。

パラメーター	説明
接続失敗時のリトライ時間	タスク開始後に DTS が失敗した接続を再試行する時間。範囲：10～1440 分。デフォルト：720 分。30 分以上に設定してください。複数のタスクが同じソースまたはターゲットデータベースを共有する場合、最も短い再試行時間が優先されます。
その他の問題のリトライ時間	DTS が失敗した DDL または DML 操作をリトライする時間。範囲: 1～1440 分。デフォルト: 10 分。10 分以上に設定することを推奨します。この値は [失敗した接続のリトライ時間] 未満である必要があります。
フルデータ同期のスロットリング有効化	完全データ同期中の読み取りおよび書き込みロードを制限します。[ソースデータベースへの QPS (クエリ/秒)]、[完全データ移行の RPS]、および [完全移行のデータ移行速度 (MB/s)] を設定します。[完全データ同期] が選択されている場合にのみ使用できます。
増分データ同期のスロットリング有効化	増分同期中の書き込みロードを制限します。[増分同期の RPS] と [増分同期のデータ同期速度 (MB/s)] を設定します。
環境タグ	DTS インスタンスを環境 (例：本番またはテスト) で識別するためのオプションのタグ。
ETL の設定	[はい] を選択して ETL (抽出·変換·書き出し) を有効にしてデータ処理文を入力するか、[いいえ] を選択してスキップします。詳細は ETL とは
監視とアラート	タスクの失敗またはレイテンシーしきい値の超過に対するアラートを設定するには、[はい] を選択します。アラートのしきい値と通知設定を設定します。詳細については、「モニタリングとアラートの設定」をご参照ください。

ステップ 6：データ検証の設定 (オプション)

[次のステップ：データ検証] をクリックしてデータ検証を設定します。詳細については、「データ検証タスクを設定する」をご参照ください。

ステップ 7：データベースとテーブルフィールドの設定 (オプション)

[次へ: データベースとテーブルのフィールドを設定] をクリックして、各テーブルの AnalyticDB for MySQL V3.0 固有の設定を構成します。

このステップは、[スキーマ同期] が選択されている場合にのみ利用可能です。すべてのテーブルを表示および変更するには、[定義ステータス] を [すべて] に設定します。

パラメーター	説明
タイプ	AnalyticDB for MySQL V3.0 のテーブルタイプ。
主キー列	テーブルのプライマリキー列。複合プライマリキーをサポートします。
分散キー	テーブルの分散キー。プライマリキー列の 1 つ以上である必要があります。
パーティションキーパーティショニングルールパーティションのライフサイクル	テーブルのパーティション分割設定。詳細については、「CREATE TABLE」をご参照ください。

ステップ 8：タスクの保存と事前チェックの実行

このタスク構成の API パラメーターをプレビューするには、[次へ: タスク設定を保存して事前チェック] にポインターを合わせ、[OpenAPI パラメーターのプレビュー] をクリックします。
[次へ: タスク設定を保存して事前チェック] をクリックすると、タスク設定が保存され、事前チェックが開始されます。

事前チェックが失敗した場合：

失敗した各項目の横にある[詳細を表示]をクリックし、問題を修正して、事前チェックを再実行します。
無視できるアラートがアイテムに表示された場合は、[アラート詳細の確認]、[無視]、[OK]、[再度事前チェック] の順にクリックします。アラートを無視すると、データの不整合が発生する可能性があります。

ステップ 9：インスタンスの購入

[成功率] が [100%] に達するのを待ってから、[次へ：インスタンスの購入] をクリックします。
購入ページで、以下を設定します。

パラメーター	説明
課金方法	サブスクリプション：固定期間に対して前払いします。長期利用に適しており、コスト効率が高いです。従量課金：時間単位で課金されます。短期利用に適しています。インスタンスが不要になった場合は、不要な課金を避けるためにリリースしてください。
リソースグループ設定	同期インスタンスが属するリソースグループです。デフォルト：デフォルトのリソースグループ。詳細については、「Resource Management とは
インスタンスクラス	同期速度のレベルです。詳細については、「データ同期インスタンスのインスタンスクラス」をご参照ください。
サブスクリプション期間	サブスクリプションの課金方法で利用可能です。選択肢：1～9 か月、1 年、2 年、3 年、5 年。

[Data Transmission Service (従量課金) 利用規約] をお読みいただき、選択します。
[購入して開始]をクリックし、次に確認ダイアログボックスで [OK] をクリックします。

タスクがタスクリストに表示されます。そこで進行状況を監視します。