Azure Data Factory を使用して Azure Cosmos DB for NoSQL のデータをコピーおよび変換する

適用対象: Azure Data Factory Azure Synapse Analytics

この記事では、Azure Data Factory の Copy アクティビティを使って Azure Cosmos DB for NoSQL との間でデータをコピーする方法、および Data Flow を使って Azure Cosmos DB for NoSQL のデータを変換する方法について説明します。 詳細については、Azure Data Factory および Azure Synapse Analytics の概要記事を参照してください。

サポートされる機能

この Azure Cosmos DB for NoSQL コネクタは、次の機能についてサポートされています。

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

Copy アクティビティの場合、この Azure Cosmos DB for NoSQL コネクタは次のことをサポートします。

• Azure リソース認証にキー、サービス プリンシパル、マネージド ID を使って、Azure Cosmos DB for NoSQL との間でデータをコピーします。
• 挿入または upsert として Azure Cosmos DB に書き込みます。
• JSON ドキュメントをインポートおよびエクスポートしたり、表形式データセットに、または表形式データセットからデータをコピーしたりします。 例としては、SQL データベースや CSV ファイルなどがあります。 JSON ファイルまたは他の Azure Cosmos DB コレクションをコピー先またはコピー元としてドキュメントをそのままコピーするには、「JSON ドキュメントのインポートとエクスポート」を参照してください。

Data Factory および Synapse パイプラインは、Azure Cosmos DB に書き込むときに最適なパフォーマンスを提供できるように、Azure Cosmos DB バルク エグゼキューター ライブラリと統合されます。

はじめに

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

• データ コピー ツール
• Azure ポータル
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure Resource Manager テンプレート

UI を使用して Azure Cosmos DB のリンク サービスを作成する

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

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

   • Azure Data Factory
   • Azure Synapse

   

2. Azure Cosmos DB for NoSQL を検索して、Azure Cosmos DB for NoSQL コネクタを選びます。

   

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

   

コネクタの構成の詳細

以下のセクションでは、Azure Cosmos DB for NoSQL に固有のエンティティの定義に使用できるプロパティについて詳しく説明します。

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

Azure Cosmos DB for NoSQL コネクタでは、次の認証の種類がサポートされています。 詳細については、対応するセクションをご覧ください。

• キー認証
• サービス プリンシパルの認証
• システム割り当てマネージド ID 認証
• ユーザー割り当てマネージド ID 認証

キー認証

例

{
    "name": "CosmosDbSQLAPILinkedService",
    "properties": {
        "type": "CosmosDb",
        "typeProperties": {
            "connectionString": "AccountEndpoint=<EndpointUrl>;AccountKey=<AccessKey>;Database=<Database>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

例: アカウント キーを Azure Key Vault に格納する

{
    "name": "CosmosDbSQLAPILinkedService",
    "properties": {
        "type": "CosmosDb",
        "typeProperties": {
            "connectionString": "AccountEndpoint=<EndpointUrl>;Database=<Database>",
            "accountKey": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<secretName>" 
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

サービス プリンシパルの認証

サービス プリンシパル認証を使用するには、次の手順に従います。

1. Microsoft ID プラットフォームにアプリケーションを登録する。 方法については、「クイック スタート: Microsoft ID プラットフォームにアプリケーションを登録する」を参照してください。 これらの値を記録しておきます。リンクされたサービスを定義するときに使います。

   • アプリケーション ID
   • アプリケーション キー
   • テナント ID

2. サービス プリンシパルに適切なアクセス許可を付与します。 Azure Cosmos DB でのアクセス許可の動作例については、ファイルとディレクトリでのアクセス制御リストに関する記事をご覧ください。 具体的には、ロール定義を作成し、サービス プリンシパル オブジェクト ID を使用してロールをサービス プリンシパルに割り当てます。

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

例: サービス プリンシパル キー認証の使用

サービス プリンシパル キーを Azure Key Vault に格納することもできます。

{
    "name": "CosmosDbSQLAPILinkedService",
    "properties": {
        "type": "CosmosDb",
        "typeProperties": {
            "accountEndpoint": "<account endpoint>",
            "database": "<database name>",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredentialType": "ServicePrincipalKey",
            "servicePrincipalCredential": {
                "type": "SecureString",
                "value": "<service principal key>"
            },
            "tenant": "<tenant info, e.g. microsoft.onmicrosoft.com>" 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

例: サービス プリンシパル認証の使用

{
    "name": "CosmosDbSQLAPILinkedService",
    "properties": {
        "type": "CosmosDb",
        "typeProperties": {
            "accountEndpoint": "<account endpoint>",
            "database": "<database name>", 
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredentialType": "ServicePrincipalCert",
            "servicePrincipalCredential": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<AKV reference>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<certificate name in AKV>" 
            },
            "tenant": "<tenant info, e.g. microsoft.onmicrosoft.com>" 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

システム割り当てマネージド ID 認証

データ ファクトリまたは Synapse パイプラインは、特定のサービス インスタンスを表す、Azure リソースのシステム割り当てマネージド ID に関連付けることができます。 独自のサービス プリンシパルを使うのと同様に、Azure Cosmos DB 認証にこのマネージド ID を直接使用できます。 これにより、この指定されたリソースは、Azure Cosmos DB のインスタンスにアクセスしてデータをコピーできます。

Azure リソースのシステム割り当てマネージド ID 認証を使用するには、次の手順に従います。

1. サービスと共に生成されたマネージド ID オブジェクト ID の値をコピーして、システム割り当てマネージド ID 情報を取得します。

2. システム割り当てマネージド ID に適切なアクセス許可を付与します。 Azure Cosmos DB でのアクセス許可の動作例については、ファイルとディレクトリでのアクセス制御リストに関する記事をご覧ください。 具体的には、ロールの定義を作成し、そのロールをシステム割り当てマネージド ID に割り当てます。

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

例:

{
    "name": "CosmosDbSQLAPILinkedService",
    "properties": {
        "type": "CosmosDb",
        "typeProperties": {
            "accountEndpoint": "<account endpoint>",
            "database": "<database name>",
            "subscriptionId": "<subscription id>",
            "tenantId": "<tenant id>",
            "resourceGroup": "<resource group>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

ユーザー割り当てマネージド ID 認証

データ ファクトリまたは Synapse パイプラインは、特定のサービス インスタンスを表す、ユーザー割り当てマネージド ID に関連付けることができます。 独自のサービス プリンシパルを使うのと同様に、Azure Cosmos DB 認証にこのマネージド ID を直接使用できます。 これにより、この指定されたリソースは、Azure Cosmos DB のインスタンスにアクセスしてデータをコピーできます。

Azure リソースのユーザー割り当てマネージド ID 認証を使用するには、次の手順に従います。

1. 1 つ以上のユーザー割り当てマネージド ID を作成し、ユーザー割り当てのマネージド ID に適切なアクセス許可を付与します。 Azure Cosmos DB でのアクセス許可の動作例については、ファイルとディレクトリでのアクセス制御リストに関する記事をご覧ください。 具体的には、ロールの定義を作成し、そのロールをユーザー割り当てマネージド ID に割り当てます。

2. 1 つ以上のユーザー割り当てマネージド ID をデータ ファクトリに割り当てて、ユーザー割り当てマネージド ID ごとに資格情報を作成します。

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

例:

{
    "name": "CosmosDbSQLAPILinkedService",
    "properties": {
        "type": "CosmosDb",
        "typeProperties": {
            "accountEndpoint": "<account endpoint>",
            "database": "<database name>",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            },
            "subscriptionId": "<subscription id>",
            "tenantId": "<tenant id>",
            "resourceGroup": "<resource group>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

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

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

Azure Cosmos DB for NoSQL データセットでは、次のプロパティがサポートされます。

"DocumentDbCollection" 型のデータセットを使用する場合、コピーおよび検索のアクティビティの下位互換性のためにそのままサポートされています。Data Flow ではサポートされていません。 今後は新しいモデルを使用することをお勧めします。

例

{
    "name": "CosmosDbSQLAPIDataset",
    "properties": {
        "type": "CosmosDbSqlApiCollection",
        "linkedServiceName":{
            "referenceName": "<Azure Cosmos DB linked service name>",
            "type": "LinkedServiceReference"
        },
        "schema": [],
        "typeProperties": {
            "collectionName": "<collection name>"
        }
    }
}

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

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

ソースとしての Azure Cosmos DB for NoSQL

Azure Cosmos DB for NoSQL からデータをコピーするには、Copy アクティビティの source の種類を DocumentDbCollectionSource に設定します。

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

"DocumentDbCollectionSource" 型のソースを使用する場合、下位互換性のためにそのままサポートされます。 Azure Cosmos DB からデータをコピーする豊富な機能を提供する新しいモデルを使用することをお勧めします。

例

"activities":[
    {
        "name": "CopyFromCosmosDBSQLAPI",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Cosmos DB for NoSQL input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "CosmosDbSqlApiSource",
                "query": "SELECT c.BusinessEntityID, c.Name.First AS FirstName, c.Name.Middle AS MiddleName, c.Name.Last AS LastName, c.Suffix, c.EmailPromotion FROM c WHERE c.ModifiedDate > \"2009-01-01T00:00:00\"",
                "preferredRegions": [
                    "East US"
                ]
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Azure Cosmos DB からデータをコピーする場合、JSON ドキュメントをそのままエクスポートする場合を除き、Copy アクティビティでマッピングを指定するのがベスト プラクティスです。 サービスでは、アクティビティで指定したマッピングが優先されます。行に列の値が含まれていない場合、列の値には null 値が指定されます。 マッピングを指定しない場合、サービスはデータの最初の行を使用してスキーマを推測します。 最初の行に完全なスキーマが含まれていないと、アクティビティ操作の結果で一部の行が欠落します。

シンクとしての Azure Cosmos DB for NoSQL

Azure Cosmos DB for NoSQL にデータをコピーするには、Copy アクティビティの sink の種類を DocumentDbCollectionSink に設定します。

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

"DocumentDbCollectionSink" 型のソースを使用する場合、下位互換性のためにそのままサポートされます。 Azure Cosmos DB からデータをコピーする豊富な機能を提供する新しいモデルを使用することをお勧めします。

例

"activities":[
    {
        "name": "CopyToCosmosDBSQLAPI",
        "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": "CosmosDbSqlApiSink",
                "writeBehavior": "upsert"
            }
        }
    }
]

スキーマ マッピング

Azure Cosmos DB から表形式のシンク、あるいはその逆の方向でデータをコピーするには、スキーマ マッピングに関するセクションを参照してください。

Mapping Data Flow のプロパティ

マッピング データフローでデータを変換する場合は、Azure Cosmos DB でコレクションの読み取りと書き込みを行うことができます。 詳細については、マッピング データ フローのソース変換とシンク変換に関する記事をご覧ください。

ソース変換

Azure Cosmos DB に固有の設定は、ソース変換の [Source Options](ソース オプション) タブにあります。

Include system columns (システム列を含める): true にすると、Azure Cosmos DB からのデータ フローのメタデータに id、_ts、その他のシステム列が含められます。 コレクションを更新するときは、既存の行 ID を把握できるように、これを含めることが重要となります。

ページ サイズ: クエリ結果のページあたりのドキュメント数。 既定値は "-1" で、サービスの動的ページが最大 1000 個使用されます。

スループット: このデータ フローの各実行について、Azure Cosmos DB コレクションの読み取り操作時に適用したい RU 数に対するオプションの値を設定します。 最小は 400 です。

Preferred regions (優先リージョン): このプロセスの優先読み取りリージョンを選択します。

変更フィード: true の場合は、Azure Cosmos DB 変更フィードからデータを取得します。これは、前回の実行から自動的に行われる順序でコンテナーに対する変更の永続的なレコードです。 ｔrue に設定する場合は、Infer drifted column typesと Allow schema drift を同時に true に設定することはできません。 詳細については、Azure Cosmos DB 変更フィード に関するページをご覧ください。

最初から開始する: true の場合は、最初の実行で完全なスナップショット データの初期読み込み、次に次の実行で変更されたデータをキャプチャします。 false の場合、最初の実行では初期読み込みはスキップされ、その後、次の実行で変更されたデータがキャプチャされます。 設定は、Azure Cosmos DB リファレンスの同じ設定名と一致します。 詳細については、Azure Cosmos DB 変更フィード に関するページをご覧ください。

シンク変換

Azure Cosmos DB に固有の設定は、シンク変換の [設定] タブにあります。

[Update method](更新方法) : 対象となるデータベースに対して許可される操作を指定します。 既定では、挿入のみが許可されます。 行を更新、アップサート、または削除するには、それらのアクションに対して行をタグ付けするために行の変更変換が必要になります。 更新、アップサート、削除の場合、1 つまたは複数のキー列を設定して、変更する行を決定する必要があります。

Collection action (コレクション アクション): 書き込み前にターゲット コレクションを再作成するかどうかを決定します。

• なし: コレクションに対してアクションは実行されません。
• 再作成: コレクションは削除され、再作成されます

バッチ サイズ: 各バッチで Azure Cosmos DB コレクションに書き込まれるオブジェクトの数を表す整数。 通常、既定のバッチ サイズで開始するだけで十分です。 この値をさらに調整するには、次の点に注意してください。

• Azure Cosmos DB では、1 つの要求のサイズは 2 MB に制限されます。 式は、"要求サイズ = 1 つのドキュメント サイズ * バッチ サイズ" となります。 "要求のサイズが大きすぎる" というエラーが発生する場合は、バッチ サイズの値を小さくします。
• バッチ サイズを大きくすると、サービスのスループットの向上を実現できますが、ワークロードを強化するために十分な RU を割り当てる必要があります。

パーティション キー:コレクションのパーティション キーを表す文字列を入力します。 例: /movies/title

スループット: このデータ フローの実行ごとに Azure Cosmos DB コレクションに適用する RU の数の値を設定します (省略可能)。 最小は 400 です。

Write throughput budget (書き込みスループット予算) : コレクションに割り当てられた合計スループットのうち、このデータ フロー書き込み操作に割り当てる RU を表す整数。

Lookup アクティビティのプロパティ

プロパティの詳細については、Lookup アクティビティに関するページを参照してください。

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

Azure Cosmos DB for NoSQL コネクタを使用して簡単に次のことができます。

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

スキーマに依存しないコピーを実行するには:

• データ コピー ツールを使うときに、[Export as-is to JSON files or Azure Cosmos DB collection](JSON ファイルまたは Azure Cosmos DB コレクションにそのままエクスポートする) オプションを選択します。
• アクティビティの作成を使用する場合は、ソースまたはシンクの対応するファイル ストアで JSON 形式を選択します。

リレーショナル データベースから Azure Cosmos DB に移行する

SQL Server などのリレーショナル データベースから Azure Cosmos DB に移行する場合、Copy アクティビティを使うと、ソースの表形式のデータを Azure Cosmos DB のフラット化された JSON ドキュメントに簡単にマッピングできます。 たとえば、1 つの JSON ドキュメント内に関連するすべてのサブ項目を埋め込んでデータを非正規化するために、データ モデルを再設計し、Azure Cosmos DB のデータ モデリングに従って NoSQL ユース ケースに合わせて最適化することもできます。 そのような場合は、Copy アクティビティを使用してこれを行う方法に関するチュートリアルを含むこちらの記事を参照してください。

Azure Cosmos DB の変更フィード

Azure Data Factory は、マッピング データ フロー ソース変換で有効にすることで、Azure Cosmos DB change feedからデータを取得できます。 このコネクタ オプションを使用すると、変換されたデータを選択した変換先データセットに読み込む前に、変更フィードを読み取り、変換を適用できます。 変更フィードを読み取り、カスタム変換を記述するために Azure functions を使用する必要はありません。 このオプションを使用すると、1 つのコンテナーから別のコンテナーにデータを移動したり、目的に合わせて変更フィード ドリブン素材ビューを準備したり、変更フィードに基づいてコンテナーのバックアップや回復を自動化したりできます。また、Azure Data Factory の視覚的なドラッグ アンド ドロップ機能を使用して、その他の多くのユース ケースを実現できます。

チェックポイントを ADF が記録して最後の実行から変更データを自動的に取得できるようにするために、パイプラインとアクティビティ名は変更しないようにしてください。 パイプライン名またはアクティビティ名を変更すると、チェックポイントがリセットされます。これにより、次回の実行時に最初から変更を開始したり、変更を取得したりすることができます。

パイプラインをデバッグすると、この機能は同じように動作します。 デバッグ実行中にブラウザーを更新すると、チェックポイントがリセットされることに注意してください。 デバッグ実行のパイプライン結果に問題がなければ、パイプラインの発行とトリガーに進むことができます。 最初に発行されたパイプラインをトリガーした時点では、最初から自動的に再起動されるか、またはその後からの変更が取得されます。

[モニター] セクションでは、常にパイプラインを再実行できます。 この場合、変更されたデータは、選択したパイプライン実行の前のチェックポイントから常にキャプチャされます。

さらに、Azure Cosmos DB 分析ストアでは、NoSQL 用 Azure Cosmos DB API と Mongo DB 用 Azure Cosmos DB API (パブリック プレビュー) で変更データ キャプチャ (CDC) がサポートされるようになりました。 Azure Cosmos DB 分析ストアを使用すると、分析ストアから変更 (挿入、更新、削除) されたデータの継続的かつ増分のフィードを効率的に使用できます。

関連するコンテンツ

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