MongoDB から Kafka へのデータ同期 - Data Transmission Service - Alibaba Cloud ドキュメントセンター

Data Transmission Service (DTS) を使用すると、ApsaraDB for MongoDB から Message Queue for Apache Kafka インスタンスに変更データをストリーミングできます。以下の手順では、MongoDB のレプリカセットをソースとし、Kafka インスタンスをデスティネーションとする同期タスクの作成方法を説明します。

このタスクで扱う内容：

対応アーキテクチャ：レプリカセットおよびシャードクラスター
増分同期方式：oplog (推奨) および変更ストリーム
データフォーマット：Canal JSON、Kafka トピックへの配信
同期範囲：コレクションレベル (完全データ同期および増分データ同期をサポート)
課金: 全量データ同期は無料、増分データ同期は有料です。

前提条件

作業を開始する前に、次の項目が満たされていることを確認してください。

Message Queue for Apache Kafka インスタンスが作成されていること
データを受信するためのトピックが、デスティネーション Kafka インスタンスに作成されていること
(シャードクラスターのみ) すべてのシャードノードに対してエンドポイントが申請済みであること。すべてのシャードノードで同じアカウントパスワードとエンドポイントを共有する必要があります。詳細については、「シャードのエンドポイントの申請」をご参照ください。

サポートされるソースおよびデスティネーションのデータベースバージョンについては、「同期ソリューション概要」をご参照ください。

制限

同期タスクを作成する前に、次の制限事項を確認してください。

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

ソースサーバーには十分なアウトバウンド帯域幅が必要です。帯域幅が不足すると、同期速度が低下します。
コレクションに名前マッピングを設定する場合、1 つのタスクで同期できるコレクションは最大 1,000 個です。この制限を超えると、リクエストエラーが発生します。1,000 個を超えるコレクションを同期するには、複数のタスクを作成するか、名前マッピングを使用せずにデータベース全体を同期してください。
シャードクラスターをソースとする場合、同期する各コレクションの _id フィールドは一意である必要があります。一意でない場合、データの不整合が発生する可能性があります。
シャードクラスターをソースとする場合、mongos ノードの数は 10 を超えることはできません。また、インスタンスに孤立ドキュメントが含まれていてはなりません。孤立ドキュメントを削除する方法については、「FAQ トピック」をご参照ください。
スタンドアロン ApsaraDB for MongoDB インスタンス、Azure Cosmos DB for MongoDB クラスター、および Amazon DocumentDB エラスティッククラスターは、ソースとしてサポートされていません。
Data Transmission Service (DTS) は、SRV エンドポイント経由で MongoDB に接続できません。
oplog を有効にして少なくとも 7 日間のログデータを保持するか、あるいは変更ストリームを有効にして DTS が過去 7 日間のデータ変更をサブスクライブできるようにする必要があります。どちらの条件も満たされていない場合、DTS はソースの変更を取得できず、サービスレベル契約 (SLA) の保証対象外となり、データ損失またはデータの不整合が発生する可能性があります。
重要
- oplog を使用してソースデータベースの変更を記録します（推奨）。 - チェンジストリームには MongoDB 4.0 以降が必要です。チェンジストリームを使用する場合、双方向同期はサポートされていません。 - 非エラスティックな Amazon DocumentDB クラスターの場合、チェンジストリームを有効にし、[移行方法] を [ChangeStream] に、[アーキテクチャ] を [Sharded Cluster] に設定します。
完全データ同期中は、データベースまたはコレクションのスキーマを変更したり、ARRAY 型のデータを変更したりしないでください。
シャードクラスターをソースとする場合、同期中に次のコマンドを実行しないでください： shardCollection、reshardCollection、unshardCollection、moveCollection、movePrimary。これらのコマンドは、データの不整合を引き起こす可能性があります。
ソースデータベースがシャードクラスターアーキテクチャを使用する MongoDB インスタンスであり、ソースデータベースのバランサーがデータをバランシングしている場合、インスタンスにレイテンシーが発生する可能性があります。

その他の制限事項

コレクションレベルの同期のみがサポートされています。
admin、config、local データベースは同期できません。
10 MB を超える単一のレコードがあると、タスクが失敗します。
トランザクションは保持されません。DTS は各トランザクションを移行先で個別のレコードに変換します。
DTS タスクの実行中に移行先の Kafka インスタンスでブローカーノードが追加または削除された場合は、DTS タスクを再起動してください。
DTS がソースおよびデスティネーションインスタンスに接続できることを確認してください。たとえば、データベースインスタンスのセキュリティ設定や、セルフマネージド Kafka インスタンスの server.properties ファイル内の listeners および advertised.listeners パラメータが、DTS からのアクセスを制限していないことを確認してください。
ソースデータベースとデスティネーションデータベースの両方で CPU 使用率が 30% を下回るオフピーク時間帯に同期を実行してください。
DTS は、実行時間が [7 日間]未満の失敗したインスタンスを自動的に再試行します。移行先へのトラフィック切り替えを行う前に、同期インスタンスを停止またはリリースして、自動再開によって移行先のデータが上書きされるのを防いでください。
DTS は、移行先データベースで最後に同期されたデータのタイムスタンプとソースデータベースの現在のタイムスタンプに基づいて、増分データ同期のレイテンシーを計算します。長期間にわたってソースデータベースで更新操作が実行されない場合、同期レイテンシーが不正確になる可能性があります。データ同期タスクのレイテンシーが極端に高い場合は、ソースデータベースで更新操作を実行してレイテンシーを更新できます。
DTS タスクの実行に失敗した場合、DTS テクニカルサポートは 8 時間以内にタスクの復元を試みます。復元中、タスクが再起動され、タスクのパラメータが変更される場合があります。変更される可能性があるのは DTS タスクのパラメータのみです。データベースのパラメータは変更されません。変更される可能性があるパラメータには、「インスタンスパラメータの変更」セクションに記載されているパラメータが含まれますが、これらに限定されません。
増分同期方式として [oplog] を使用するシャードクラスターをソースとする場合、DTS はターゲットの Kafka トピックへのシャード間の書き込み順序を保証しません。
シャードクラスターをソースとする場合、完全データ同期中は MongoDB バランサーを無効にしてください。完全同期が完了し、増分データ同期が開始された後にのみ再度有効にしてください。詳細については、「ApsaraDB for MongoDB バランサーの管理」をご参照ください。

課金

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

同期タイプとサポートされる操作

[完全同期] は、選択された MongoDB コレクションからすべての既存データをターゲットの Kafka トピックにコピーします。DTS は DATABASE と COLLECTION オブジェクトをサポートしています。

[増分同期] は、完全同期の完了後に変更イベントを継続的に配信します。サポートされる操作は、増分同期方式によって異なります。

操作	Oplog	変更ストリーム
INSERT	サポートされています	サポートされています
UPDATE	サポートされています	サポートされています
DELETE	サポートされています	サポートされています
CREATE COLLECTION / INDEX	サポートされています	サポートされていません
DROP DATABASE / COLLECTION / INDEX	サポートされています	データベースとコレクションの削除のみ
RENAME COLLECTION	サポートされています	サポートされています

増分同期では、タスクの開始後に作成されたデータベースは取得されません。

必要なデータベースアカウントの権限

データベース	必要な権限
ソース ApsaraDB for MongoDB	同期するデータベース、`admin` データベース、および `local` データベースの読み取り権限

アカウントの作成手順については、「アカウント管理」をご参照ください。

説明

増分同期方法として ChangeStream を使用する場合、ソースデータベースアカウントにはインスタンス全体の Change Streams 読み取り権限 (readAnyDatabase など) が必要です。ソースがカスタムアカウントを使用する ApsaraDB for MongoDB インスタンスの場合、アカウントに admin データベースに対する読み取り権限も付与する必要があります。詳細については、「インスタンス作成時に指定する root アカウントの権限」をご参照ください。

同期タスクの作成

ステップ 1：データ同期ページへのアクセス

Data Transmission Service (DTS) コンソールまたは DMS コンソールのいずれかを使用します。

DTS コンソール

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

DMS コンソール

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

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

ステップ 2：ソースデータベースと移行先データベースの設定

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

全般

パラメータ	説明
[タスク名]	DTS タスクの名前です。DTS は自動的に名前を生成します。タスクを識別しやすいように、分かりやすい名前を指定します。名前は一意である必要はありません。

ソースデータベース

パラメータ	説明
[既存の接続を選択]	ソースインスタンスが DTS に登録されている場合は、ドロップダウンリストから選択します。DTS は残りのフィールドを自動的に入力します。それ以外の場合は、以下のフィールドを設定します。DMS コンソールでは、[DMS データベースインスタンスを選択] リストから選択します。
[データベースタイプ]	[MongoDB] を選択します。
[アクセス方法]	[Alibaba Cloud インスタンス] を選択します。
[インスタンスリージョン]	ソース ApsaraDB for MongoDB インスタンスのリージョンを選択します。
[Alibaba Cloud アカウント間でのデータレプリケーション]	ソースデータベースが現在の Alibaba Cloud アカウントに属している場合は、[いいえ] を選択します。
[アーキテクチャ]	この例では、[レプリカセット] を選択します。ソースが [シャードクラスター] の場合、[シャードアカウント] と [シャードパスワード] も入力します。
[移行方法]	増分データを同期する方法です。[Oplog] は、oplog 機能が有効な場合に推奨されます。[ChangeStream] は、ソースでチェンジストリームが有効になっている場合に使用できます。詳細については、「チェンジストリーム」をご参照ください。注：非エラスティックな Amazon DocumentDB クラスターの場合、[ChangeStream] のみがサポートされます。[アーキテクチャ] が [シャーディングクラスター] に設定されており、[移行方法] が [ChangeStream] に設定されている場合、[シャードアカウント] フィールドと [シャードパスワード] フィールドは不要です。
[インスタンス ID]	ソース ApsaraDB for MongoDB インスタンスのインスタンス ID を選択します。
[認証データベース]	アカウント認証情報を含むデータベースです。デフォルトは `admin` です。
[データベースアカウント]	ソースデータベースのアカウントです。詳細については、「必要なデータベースアカウント権限」をご参照ください。
[データベースパスワード]	データベースアカウントのパスワードです。
[暗号化]	接続の暗号化タイプを指定します。選択肢は [非暗号化]、[SSL 暗号化]、または [Mongo Atlas SSL] です。使用可能なオプションは、[アクセス方法] と [アーキテクチャ] の値によって異なります。注: [アーキテクチャ] が [シャーディングクラスター] で、[移行方法] が [Oplog] の場合、SSL 暗号化は利用できません。自己管理型の MongoDB で [レプリカセット] アーキテクチャを使用し、[SSL 暗号化] を選択した場合は、接続を検証するために CA 証明書をアップロードしてください。

移行先データベース

パラメータ	説明
[既存の接続を選択]	移行先 Kafka インスタンスが DTS に登録されている場合は、ドロップダウンリストから選択します。それ以外の場合は、以下のフィールドを設定します。
[データベースタイプ]	[Kafka] を選択します。
[アクセス方法]	[Alibaba Cloud インスタンス] を選択します。
[インスタンスリージョン]	移行先 Kafka インスタンスのリージョンを選択します。
[Kafka インスタンス ID]	移行先 Kafka インスタンスの ID を選択します。
[暗号化]	セキュリティ要件に応じて、[非暗号化] または [SCRAM-SHA-256] を選択します。
[トピック]	同期されたデータを受信するトピックを選択します。
[Kafka スキーマレジストリの使用]	Kafka Schema Registry は、Avro スキーマを保存および取得するための RESTful メタデータサービスです。スキップするには [いいえ] を選択するか、[はい]アラート通知設定を選択して、Schema Registry の URL または IP アドレスを指定します。

ステップ 3：接続のテスト

ページの下部にある[接続をテストして続行]をクリックします。

DTS は、許可されている場合、ソースデータベースと移行先データベースのセキュリティ設定に DTS サーバーの CIDR ブロックを自動的に追加します。手動設定については、「DTS サーバーの CIDR ブロックの追加」をご参照ください。

アクセス方法として [Alibaba Cloud Instance] を使用しない自己管理型データベースの場合は、[DTS サーバーの CIDR ブロック] ダイアログボックスで [接続テスト] をクリックします。

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

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

パラメータ	説明
[同期タイプ]	デフォルトで [増分データ同期] が選択されています。オプションで [完全データ同期] も選択できます。[スキーマ同期] は利用できません。完全同期が有効な場合、DTS は増分同期を開始する前に、まずすべての履歴データをコピーします。
[競合するテーブルの処理モード]	[事前チェックとエラー報告]: 移行先に移行元と同じ名前のコレクションが含まれている場合、事前チェックは失敗します。オブジェクト名のマッピングを使用して、名前の競合を解決します。[エラーを無視して続行]: 名前の競合チェックをスキップします。移行先のレコードに移行元のレコードと同じプライマリキーまたは一意キーがある場合、移行先のレコードが保持されます。警告このオプションはデータの不整合を引き起こす可能性があります。
[Kafka のデータフォーマット]	Canal JSON のみがサポートされています。
[Kafka データ圧縮フォーマット]	Kafka に書き込まれるデータの圧縮形式です。オプション: [LZ4] (デフォルト — 低い圧縮率、高速)、[GZIP] (高い圧縮率、低速、高い CPU 使用率)、[Snappy] (バランス型)。
[Kafka パーティションへのデータ配信ポリシー]	要件に基づいてパーティションルーティングポリシーを選択します。
[メッセージ確認メカニズム]	要件に基づいてメッセージ確認メカニズムを選択します。
[DDL 情報を格納するトピック]	DDL イベントを格納するトピックを選択します。空白のままにすると、DDL イベントはデータイベントと同じトピックに書き込まれます。
[移行先インスタンスでのオブジェクト名の大文字小文字]	移行先のデータベース名およびコレクション名の大文字/小文字の区別を制御します。デフォルトは [DTS デフォルトポリシー] です。詳細については、「移行先インスタンスのオブジェクト名の大文字/小文字の区別を指定する」をご参照ください。
[ソースオブジェクト]	[ソースオブジェクト] セクションからオブジェクトを選択し、矢印アイコンをクリックして [選択済みオブジェクト] に移動します。コレクションレベルの選択のみサポートされています。

ステップ 5：詳細設定の構成

[次へ: 詳細設定] をクリックし、以下のパラメーターを設定します。

パラメータ	説明
[タスクスケジューリング専用クラスター]	デフォルトでは、DTS は共有クラスターでタスクをスケジュールします。より高い安定性を得るには、専用クラスターを購入できます。
[接続失敗時の再試行時間]	ソースまたは移行先に到達できない場合に DTS が再試行する時間です。範囲：10 ～ 1440 分。デフォルト：720 分。このパラメータは 30 より大きい値に設定することを推奨します。同じデータベースを共有する複数のタスクに異なる再試行時間を指定した場合、最短の値が有効になります。DTS は再試行時間に対して課金します。
[その他の問題の再試行時間]	DDL または DML 操作が失敗した場合に DTS が再試行する時間を指定します。範囲: 1～1440 分。デフォルト: 10 分。少なくとも 10 分に設定し、[接続失敗時の再試行時間]より短くする必要があります。
[更新後にドキュメント全体を取得]	[移行方法] が [ChangeStream] の場合にのみ利用可能です。 [はい]: 更新ごとにドキュメント全体を同期するため、ソースの負荷が増加し、レイテンシーが発生する場合があります。 DTS がドキュメント全体を取得できない場合、更新されたフィールドのみが送信されます。 [いいえ]: 変更されたフィールドのみを同期します。
[完全データ同期のスロットリングを有効化]	有効にすると、[ソースデータベースへの QPS]、[全量データ移行の RPS]、および[全量移行のデータ移行速度 (MB/s)] を設定して、移行先の負荷を軽減できます。[全量データ同期] が選択されている場合にのみ利用可能です。
[同期対象テーブルのプライマリキー_idのデータ型が1種類のみ]	[フルデータ同期] が選択されている場合にのみ利用可能です。[はい]: DTS はフル同期中に `_id` フィールドのデータ型のスキャンをスキップします。[いいえ]: DTS は `_id` フィールドのデータ型をスキャンします。
[増分データ同期のスロットリングを有効化]	有効にした場合、移行先の負荷を軽減するために、[増分データ同期の RPS] と [増分データ同期の速度 (MB/s)] を設定します。
[環境タグ]	インスタンスを識別するためのオプションのタグです。
[ETL の設定]	有効にすると、抽出、変換、ロード (ETL) ロジックが適用されます。コードエディタにデータ処理ステートメントを入力します。詳細については、「データ移行またはデータ同期タスクでの ETL の設定」をご参照ください。
[監視とアラート]	有効にすると、アラートしきい値と通知設定を指定します。タスクが失敗した場合、または同期レイテンシーがしきい値を超えた場合に、DTS はアラート連絡先に通知します。詳細については、「監視とアラートの設定」をご参照ください。

ステップ 6：事前チェックの実行

[次へ：タスク設定の保存と事前チェック] をクリックします。

この設定の API パラメーターをプレビューするには、先に進む前にボタンにカーソルを合わせ、[OpenAPI パラメーターのプレビュー] をクリックします。

事前チェックに合格するまで、タスクは開始できません。

事前チェックが失敗した場合は、失敗した各項目の横にある[詳細の表示]をクリックして問題を解決し、[再度事前チェック]をクリックします。

アイテムに対してアラートが発生した場合、続行する前に無視できないアラートを修正してください。無視できるアラートについては、[アラート詳細の確認] をクリックし、ダイアログで [無視] をクリックして確認した後、[再事前チェック] をクリックします。アラートを無視すると、データの不整合が発生する可能性があります。

ステップ 7：インスタンスの購入と開始

[成功率] が [100%] に達したら、[次へ: インスタンスの購入] をクリックします。

[購入] ページで、次のパラメーターを設定します。

パラメータ	説明
[課金方法]	[サブスクリプション]: 前払いで、長期間のご利用に費用対効果が高くなります。従量課金: 時間単位で課金され、短期間のご利用に適しています。不要になったインスタンスを解放すると、課金が停止されます。
[リソースグループ設定]	同期インスタンスのリソースグループ。デフォルト: [デフォルトのリソースグループ]。詳細については、「リソース管理とは」をご参照ください。
[インスタンスクラス]	同期速度の階層です。詳細については、「データ同期インスタンスのインスタンスクラス」をご参照ください。
[サブスクリプション期間]	[サブスクリプション] 課金で利用可能です。選択肢: 1～9 か月、1 年、2 年、3 年、または 5 年。

[データ伝送サービス (従量課金) サービス規約]を読み、同意してください。
[購入して開始] をクリックし、確認ダイアログで [OK] をクリックします。

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

コレクションとトピックのマッピング設定

デフォルトでは、すべてのコレクションは宛先データベースの設定で選択したトピックに書き込まれます。特定のコレクションを別のトピックにルーティングするには、次の手順に従います。

[選択したオブジェクト] エリアで、コレクションレベルの宛先トピック名にマウスカーソルを合わせます。
トピック名の横にある[編集]をクリックします。

[テーブルの編集] ダイアログで、次の設定を構成します。

パラメーター	説明
[宛先トピック名]	このコレクションの宛先トピック。トピックは Kafka インスタンスに存在する必要があります。変更すると、データは新しいトピックに書き込まれます。デフォルト：宛先データベースの設定時に選択したトピック。
[フィルター条件]	オプションの行フィルター。詳細については、「フィルター条件の設定」をご参照ください。
[パーティション数]	トピックにデータを書き込む際のパーティション数。

[OK] をクリックします。

データ配信例

MongoDB からの各増分変更は、Canal JSON メッセージとしてシリアル化され、設定済みの Kafka トピックに配信されます。メッセージ構造は、増分同期方式と更新の設定によって異なります。

シナリオの選択

目標	設定
完全な DDL をサポートする低レイテンシー同期	[移行方式] を [Oplog] に設定します (シナリオ 1)
ドキュメントの部分更新を伴う変更ストリーム	[移行方式] を [ChangeStream] に、更新後にドキュメント全体を取得するを [いいえ] に設定します (シナリオ 2)
更新ごとにフルドキュメントを含む変更ストリーム	[移行方式] を [ChangeStream] に、更新後にドキュメント全体を取得するを [はい] に設定します (シナリオ 3)

Canal JSON メッセージのフィールド

3 つのシナリオはすべて、同じトップレベルの Canal JSON エンベロープを使用します。すべてのメッセージに次のフィールドが含まれます。

フィールド	タイプ	説明
`database`	string	ソースデータベース名
`table`	string	ソースコレクション名
`type`	string	操作タイプ：`INSERT`、`UPDATE`、`DELETE`、または `DDL`
`isDdl`	boolean	DDL イベント (コレクションの削除、名前の変更) の場合は `true`、DML イベントの場合は `false` です。
`es`	number	ソースデータベースのイベントタイムスタンプ (Unix ミリ秒)
`ts`	number	DTS がイベントを処理した時点のタイムスタンプ (Unix ミリ秒)
`id`	number	DTS の内部イベント ID
`pkNames`	array	プライマリキーのフィールド名 (通常は `["_id"]`)
`data`	array	操作後のドキュメントデータ。部分更新 (Oplog またはフルドキュメント取得なしの ChangeStream) の場合、MongoDB の更新演算子 (`$set`、`$unset`) を使用して変更されたフィールドのみが含まれます。DDL イベントの場合は `null` です。
`old`	array	更新前のドキュメントの状態：`_id` フィールドのみを含みます。INSERT および DDL イベントの場合は `null` です。
`sql`	object or null	DDL ステートメントの詳細 (`isDdl: true` の場合)。DML イベントの場合は `null` です。
`gtid`	null	MongoDB では適用されません。MySQL との互換性のために予約されています。
`mysqlType`	null	MongoDB では適用されません。
`serverId`	null	MongoDB では適用されません。
`sqlType`	null	MongoDB では適用されません。

シナリオ 1：Oplog

[移行方式] を [Oplog] に設定します。

ソースの変更タイプ	ソースステートメント	宛先トピックが受信したデータ
`insert`	`db.kafka_test.insert({"cid":"a","person":{"name":"testName","age":NumberInt(18),"skills":["database","ai"]}})`	以下の例を参照してください
`update $set`	`db.kafka_test.update({"cid":"a"},{$set:{"person.age":NumberInt(20)}})`	以下の例を参照してください
`update $set new field`	`db.kafka_test.update({"cid":"a"},{$set:{"salary":100}})`	以下の例を参照してください
`update $unset (フィールドの削除)`	`db.kafka_test.update({"cid":"a"},{$unset:{"salary":1}})`	以下の例を参照してください
`delete`	`db.kafka_test.deleteOne({"cid":"a"})`	以下の例を参照してください
`ddl drop`	`db.kafka_test.drop()`	以下の例を参照してください

データを表示 (クリックして展開)

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 18
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741847972000,
    "gtid": null,
    "id": 174184797200000****,
    "isDdl": false,
    "mysqlType": null,
    "old": null,
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741847973438,
    "type": "INSERT"
}

データを表示 (クリックして展開)

{
    "data": [{
        "$set": {
            "person.age": 20
        }
    }],
    "database": "kafkadb",
    "es": 1741848051000,
    "gtid": null,
    "id": 174184805100000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848051984,
    "type": "UPDATE"
}

データを表示 (クリックして展開)

{
    "data": [{
        "$set": {
            "salary": 100.0
        }
    }],
    "database": "kafkadb",
    "es": 1741848146000,
    "gtid": null,
    "id": 174184814600000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848147734,
    "type": "UPDATE"
}

データを表示 (クリックして展開)

{
    "data": [{
        "$unset": {
            "salary": true
        }
    }],
    "database": "kafkadb",
    "es": 1741848207000,
    "gtid": null,
    "id": 174184820700000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848208186,
    "type": "UPDATE"
}

データを表示 (クリックして展開)

{
    "data": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "database": "kafkadb",
    "es": 1741848289000,
    "gtid": null,
    "id": 174184828900000****,
    "isDdl": false,
    "mysqlType": null,
    "old": null,
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848289798,
    "type": "DELETE"
}

データを表示 (クリックして展開)

{
    "data": null,
    "database": "kafkadb",
    "es": 1741847893000,
    "gtid": null,
    "id": 1741847893000000005,
    "isDdl": true,
    "mysqlType": null,
    "old": null,
    "pkNames": null,
    "serverId": null,
    "sql": {
        "drop": "kafka_test"
    },
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741847893760,
    "type": "DDL"
}

シナリオ 2：ChangeStream — 更新されたフィールドのみ

[移行方式] を [ChangeStream] に設定します。[更新後にドキュメント全体を取得する] を [いいえ] に設定します。

挿入および削除メッセージはシナリオ 1 と同じです。更新メッセージには、変更されたフィールドのみが含まれます。

データを表示 (クリックして展開)

{
    "data": [{
        "$set": {
            "person.age": 20
        }
    }],
    "database": "kafkadb",
    "es": 1741848051000,
    "gtid": null,
    "id": 174184805100000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848052912,
    "type": "UPDATE"
}

データを表示 (クリックして展開)

{
    "data": [{
        "$unset": {
            "salary": true
        }
    }],
    "database": "kafkadb",
    "es": 1741848207000,
    "gtid": null,
    "id": 174184820700000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848209142,
    "type": "UPDATE"
}

シナリオ 3：ChangeStream — 更新時にフルドキュメント

[移行方式] を [ChangeStream] に設定します。[更新後にドキュメント全体を取得する] を [はい] に設定します。

更新イベントは、変更されたフィールドだけでなく、変更後のフルドキュメントを配信します。

データを表示 (クリックして展開)

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 20
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741848051000,
    "gtid": null,
    "id": 174184805100000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848052219,
    "type": "UPDATE"
}

データを表示 (クリックして展開)

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 20
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "salary": 100.0,
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741848146000,
    "gtid": null,
    "id": 174184814600000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848147327,
    "type": "UPDATE"
}

データを表示 (クリックして展開)

{
    "data": [{
        "person": {
            "skills": ["database", "ai"],
            "name": "testName",
            "age": 20
        },
        "_id": {
            "$oid": "67d27da49591697476e1****"
        },
        "cid": "a"
    }],
    "database": "kafkadb",
    "es": 1741848207000,
    "gtid": null,
    "id": 174184820700000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": {
            "$oid": "67d27da49591697476e1****"
        }
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "kafka_test",
    "ts": 1741848208401,
    "type": "UPDATE"
}

特殊なケース：ChangeStream での fullDocument の欠落

ChangeStream 更新イベントの fullDocument フィールドが欠落している場合 (たとえば、ドキュメントがシャーディングされたコレクション内のシャード間を移動する場合)、配信されるメッセージは Oplog の動作 ($set または $unset 演算子を使用した部分更新) にフォールバックします。

例：fullDocument が欠落しているシャーディングされたコレクションの更新

ソースの基本データ：

use admin
db.runCommand({ enablesharding:"dts_test" })

ソースの増分変更：

use dts_test
sh.shardCollection("dts_test.cstest",{"name":"hashed"})
db.cstest.insert({"_id":1,"name":"a"})
db.cstest.updateOne({"_id":1,"name":"a"},{$set:{"name":"b"}})

データを表示 (クリックして展開)

{
    "data": [{
        "$set": {
            "name": "b"
        }
    }],
    "database": "dts_test",
    "es": 1740720994000,
    "gtid": null,
    "id": 174072099400000****,
    "isDdl": false,
    "mysqlType": null,
    "old": [{
        "_id": 1.0
    }],
    "pkNames": ["_id"],
    "serverId": null,
    "sql": null,
    "sqlType": null,
    "table": "cstest",
    "ts": 1740721007099,
    "type": "UPDATE"
}

よくある質問

タスクの作成後に Kafka データ圧縮フォーマットを変更できますか？

はい。 [「同期対象の変更」] をご参照ください。

タスクの作成後にメッセージ確認応答メカニズムを変更できますか？

はい。 [「同期対象の変更」] をご参照ください。

Data Transmission Service:ApsaraDB for MongoDB から Message Queue for Apache Kafka へのデータ同期

前提条件

制限

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

その他の制限事項

課金

同期タイプとサポートされる操作

必要なデータベースアカウントの権限

同期タスクの作成

ステップ 1：データ同期ページへのアクセス

ステップ 2：ソースデータベースと移行先データベースの設定

ステップ 3：接続のテスト

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

ステップ 5：詳細設定の構成

ステップ 6：事前チェックの実行

ステップ 7：インスタンスの購入と開始

コレクションとトピックのマッピング設定

データ配信例

シナリオの選択

Canal JSON メッセージのフィールド

シナリオ 1：Oplog

シナリオ 2：ChangeStream — 更新されたフィールドのみ

シナリオ 3：ChangeStream — 更新時にフルドキュメント

特殊なケース：ChangeStream での fullDocument の欠落

よくある質問

次のステップ