Azure Data Factory または Azure Synapse Analytics を使用して MongoDB Atlas との間でデータをコピーする

  • [アーティクル]

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータ サイエンス、リアルタイム分析、ビジネス インテリジェンス、レポートまで、あらゆるものをカバーしています。 無料で新しい試用版を開始する方法について説明します。

この記事では、Azure Data Factory または Synapse Analytics パイプラインで Copy アクティビティを使用して、MongoDB Atlas データベースからデータをコピーする方法について説明します。 この記事は、コピー アクティビティの概要を示しているコピー アクティビティの概要に関する記事に基づいています。

サポートされる機能

この MongoDB Atlas コネクタでは、次の機能がサポートされます。

サポートされる機能 IR
Copy アクティビティ (ソース/シンク) ① ②

① Azure 統合ランタイム ② セルフホステッド統合ランタイム

ソースおよびシンクとしてサポートされているデータ ストアの一覧については、「サポートされているデータ ストア」の表を参照してください。

前提条件

データ ストアがオンプレ ミスネットワーク、Azure 仮想ネットワーク、または Amazon Virtual Private Cloud 内にある場合は、それに接続するようセルフホステッド統合ランタイムを構成する必要があります。

データ ストアがマネージド クラウド データ サービスである場合は、Azure Integration Runtime を使用できます。 ファイアウォール規則で承認されている IP にアクセスが制限されている場合は、Azure Integration Runtime の IP を許可リストに追加できます。

また、Azure Data Factory のマネージド仮想ネットワーク統合ランタイム機能を使用すれば、セルフホステッド統合ランタイムをインストールして構成しなくても、オンプレミス ネットワークにアクセスすることができます。

Data Factory によってサポートされるネットワーク セキュリティ メカニズムやオプションの詳細については、「データ アクセス戦略」を参照してください。

作業の開始

パイプラインでコピー アクティビティを実行するには、次のいずれかのツールまたは SDK を使用します。

UI を使用して MongoDB Atlas のリンク サービスを作成する

次の手順を使用して、Azure portal UI で MongoDB Atlas のリンク サービスを作成します。

  1. Azure Data Factory または Synapse ワークスペースの [管理] タブに移動し、[リンク サービス] を選択して、[新規] をクリックします。

  2. "MongoDB Atlas" を検索し、MongoDB Atlas コネクタを選びます。

    Select the MongoDB Atlas connector.

  3. サービスの詳細を構成し、接続をテストして、新しいリンク サービスを作成します。

    Configure a linked service to MongoDB Atlas.

コネクタの構成の詳細

以下のセクションでは、MongoDB Atlas コネクタに固有の Data Factory エンティティを定義するために使用されるプロパティの詳細について説明します。

リンクされたサービスのプロパティ

MongoDB Atlas のリンクされたサービスでは、次のプロパティがサポートされます。

プロパティ Description 必須
type type プロパティは、次のように設定する必要があります:MongoDbAtlas はい
connectionString MongoDB Atlas 接続文字列 (例: mongodb+srv://<username>:<password>@<clustername>.<randomString>.<hostName>/<dbname>?<otherProperties>) を指定します。

接続文字列を Azure Key Vault に格納することもできます。 詳細については、「Azure Key Vault への資格情報の格納」を参照してください。		 はい
database アクセスするデータベースの名前。 はい
driverVersion MongoDB バージョン 3.6 以降をサポートする v2 にドライバーのバージョンを指定します。 詳しくは、こちらの記事をご覧ください。 いいえ
connectVia データ ストアに接続するために使用される統合ランタイム。 詳細については、「前提条件」セクションを参照してください。 指定されていない場合は、既定の Azure 統合ランタイムが使用されます。 いいえ

例:

{
    "name": "MongoDbAtlasLinkedService",
    "properties": {
        "type": "MongoDbAtlas",
        "typeProperties": {
            "connectionString": "mongodb+srv://<username>:<password>@<clustername>.<randomString>.<hostName>/<dbname>?<otherProperties>",
            "database": "myDatabase",
            "driverVersion": "<driver version>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

データセットのプロパティ

データセットの定義に使用できるセクションとプロパティの完全な一覧については、「データセットとリンクされたサービス」を参照してください。 MongoDB Atlas データセットでは、次のプロパティがサポートされます。

プロパティ Description 必須
type データセットの type プロパティは、次のように設定する必要があります:MongoDbAtlasCollection はい
collectionName MongoDB Atlas データベースのコレクション名。 はい

例:

{
    "name": "MongoDbAtlasDataset",
    "properties": {
        "type": "MongoDbAtlasCollection",
        "typeProperties": {
            "collectionName": "<Collection name>"
        },
        "schema": [],
        "linkedServiceName": {
            "referenceName": "<MongoDB Atlas linked service name>",
            "type": "LinkedServiceReference"
        }
    }
}

コピー アクティビティのプロパティ

アクティビティの定義に利用できるセクションとプロパティの完全な一覧については、パイプラインに関する記事を参照してください。 このセクションでは、MongoDB Atlas ソースおよびシンクでサポートされるプロパティの一覧を示します。

ソースとしての MongoDB Atlas

コピー アクティビティの source セクションでは、次のプロパティがサポートされます。

プロパティ Description 必須
type コピー アクティビティのソースの type プロパティは、次のように設定する必要があります:MongoDbAtlasSource はい
filter クエリ演算子を使用して選択フィルターを指定します。 コレクション内のすべてのドキュメントを返すには、このパラメーターを省略するか、空のドキュメント ({}) を渡します。 いいえ
cursorMethods.project プロジェクションのドキュメントで返されるフィールドを指定します。 一致するドキュメント内のすべてのフィールドを返すには、このパラメーターを省略します。 いいえ
cursorMethods.sort 一致するドキュメントがクエリによって返される順序を指定します。 「cursor.sort()」を参照してください。 いいえ
cursorMethods.limit サーバーが返すドキュメントの最大数を指定します。 「cursor.limit()」を参照してください。 いいえ
cursorMethods.skip スキップするドキュメントの数と、MongoDB Atlas が結果を返すときの開始位置を指定します。 「cursor.skip()」を参照してください。 いいえ
batchSize MongoDB Atlas インスタンスからの応答の各バッチで返されるドキュメントの数を指定します。 ほとんどの場合、バッチ サイズを変更しても、ユーザーまたはアプリケーションへの影響はありません。 Azure Cosmos DB では各バッチのサイズが 40 MB を超過しないように制限されていますが、これはドキュメントが batchSize の数だけ存在するときの合計サイズなので、ドキュメントのサイズが大きくなる場合はこの値を減らしてください。 いいえ
(既定値は 100)

ヒント

このサービスは、厳格モードでの BSON ドキュメントの利用をサポートしています。 フィルター クエリがシェル モードではなく厳格モードであることを確認してください。 詳細については、MongoDB のマニュアルを参照してください。

例:

"activities":[
    {
        "name": "CopyFromMongoDbAtlas",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<MongoDB Atlas input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "MongoDbAtlasSource",
                "filter": "{datetimeData: {$gte: ISODate(\"2018-12-11T00:00:00.000Z\"),$lt: ISODate(\"2018-12-12T00:00:00.000Z\")}, _id: ObjectId(\"5acd7c3d0000000000000000\") }",
                "cursorMethods": {
                    "project": "{ _id : 1, name : 1, age: 1, datetimeData: 1 }",
                    "sort": "{ age : 1 }",
                    "skip": 3,
                    "limit": 3
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

シンクとしての MongoDB Atlas

コピー アクティビティの sink セクションでは、次のプロパティがサポートされます。

プロパティ Description 必須
type Copy アクティビティのシンクの type プロパティには MongoDbAtlasSink を設定する必要があります。 はい
writeBehavior MongoDB Atlas にデータを書き込む方法について説明します。 使用可能な値は、Insert、Upsert です。

upsert の動作は、同じ _id を持つドキュメントが既に存在する場合、そのドキュメントを置き換えます。それ以外の場合は、ドキュメントを挿入します。

注: 元のドキュメントまたは列マッピングで _id が指定されていない場合は、サービスによってドキュメントの _id が自動的に生成されます。 つまり、upsert が期待どおりに動作するには、ドキュメントに ID があることを確認する必要があります。		 いいえ
(既定値は insert です)
writeBatchSize writeBatchSize プロパティにより、各バッチで書き込むドキュメントのサイズが制御されます。 パフォーマンスを向上させるには writeBatchSize の値を大きくしてみて、ドキュメントのサイズが大きい場合は値を小さくしてみます。 いいえ
(既定値は 10,000)
writeBatchTimeout タイムアウトするまでに一括挿入操作の完了を待つ時間です。許容される値は期間です。 いいえ
(既定値は 00:30:00 - 30 分)

ヒント

JSON ドキュメントをそのままインポートするには、「JSON ドキュメントをインポートまたはエクスポートする」セクションを参照してください。表形式のデータからコピーするには、「スキーマ マッピング」を参照してください。

例

"activities":[
    {
        "name": "CopyToMongoDBAtlas",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Document DB output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "MongoDbAtlasSink",
                "writeBehavior": "upsert"
            }
        }
    }
]

JSON ドキュメントのインポートとエクスポート

この MongoDB Atlas コネクタを使用すると、簡単に次を実行できます。

  • 2 つの MongoDB Atlas コレクション間でドキュメントをそのままコピーします。
  • Azure Cosmos DB、Azure Blob Storage、Azure Data Lake Store、その他のサポートされているファイルベースのストアなど、さまざまなソースから MongoDB Atlas に JSON ドキュメントをインポートします。
  • JSON ドキュメントをMongoDB Atlas コレクションからさまざまなファイル ベースのストアにエクスポートします。

このようなスキーマに依存しないコピーを実現するには、データセットの "構造" ("スキーマ" とも呼ばれる) のセクションと、コピー アクティビティでのスキーマ マッピングをスキップします。

スキーマ マッピング

MongoDB Atlas から表形式のシンクにデータをコピーする、またはシンクから MongoDB Atlas にデータをコピーするには、スキーマ マッピングに関する記事をご覧ください。

関連するコンテンツ

Copy アクティビティでソースおよびシンクとしてサポートされるデータ ストアの一覧については、サポートされるデータ ストアに関するセクションを参照してください。