.NET SDK を使用して Azure Cosmos DB for NoSQL アカウントにデータを一括インポートする

[アーティクル]
06/01/2023

適用対象: NoSQL

このチュートリアルでは、Azure Cosmos DB にデータをインポートするために必要なプロビジョニングスループット (RU/秒) を最適化する .NET コンソールアプリケーションを構築する方法について説明します。

この記事では、サンプルデータソースからデータを読み取り、Azure Cosmos DB コンテナーにインポートします。このチュートリアルでは、.NET Framework または .NET Core をターゲットとすることができる、Azure Cosmos DB .NET SDK のバージョン 3.0 以降を使用します。

このチュートリアルの内容:

Azure Cosmos DB アカウントを作成する
プロジェクトを構成する
一括サポートが有効になっている Azure Cosmos DB アカウントに接続する
同時作成操作によるデータインポートを実行する

前提条件

この記事の手順を実行する前に、以下のリソースがあることを確認してください。

アクティブな Azure アカウントアカウントがない場合、Azure 試用版にサインアップして、最大 10 件の無料 Mobile Apps を入手できます。 Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。

Azure サブスクリプション不要で、契約もなしで Azure Cosmos DB を無料で試すことができます。または、Azure Cosmos DB Free レベルのアカウントを作成して、最初の 1000 RU/秒と 25 GB のストレージを無料でご利用いただけます。また、URI https://localhost:8081 で Azure Cosmos DB エミュレーターを使用することもできます。エミュレーターで使用するキーについては、「要求の認証」を参照してください。
NET Core 3 SDK。 dotnet --version を実行すると、お使いの環境で使用可能なバージョンを確認できます。

手順 1:Azure Cosmos DB アカウントを作成する

Azure portal から Azure Cosmos DB for NoSQL アカウントを作成します。または、Azure Cosmos DB Emulator を使用してアカウントを作成することもできます。

手順 2: .NET プロジェクトを設定する

ローカルコンピューターから Windows コマンドプロンプトまたはターミナルウィンドウを開きます。次のセクションではコマンドプロンプトまたはターミナルからすべてのコマンドを実行します。次の dotnet new コマンドを実行して、bulk-import-demo という名前の新しいアプリを作成します。

dotnet new console -n bulk-import-demo

新しく作成されたアプリフォルダーにディレクトリを変更します。次を使用してアプリケーションをビルドできます。

cd bulk-import-demo
dotnet build

ビルドからの予想される出力は、次のようになります。

Restore completed in 100.37 ms for C:\Users\user1\Downloads\CosmosDB_Samples\bulk-import-demo\bulk-import-demo.csproj.
  bulk -> C:\Users\user1\Downloads\CosmosDB_Samples\bulk-import-demo \bin\Debug\netcoreapp2.2\bulk-import-demo.dll

Build succeeded.
    0 Warning(s)
    0 Error(s)

Time Elapsed 00:00:34.17

手順 3: Azure Cosmos DB パッケージを追加する

アプリケーションディレクトリ内で、dotnet add package コマンドを使用して .NET Core 用の Azure Cosmos DB クライアントライブラリをインストールします。

dotnet add package Microsoft.Azure.Cosmos

手順 4: Azure Cosmos DB アカウントの資格情報を取得する

サンプルアプリケーションは、Azure Cosmos DB アカウントに対して認証を行う必要があります。認証するには、Azure Cosmos DB アカウントの資格情報をアプリケーションに渡す必要があります。次のこれらの手順に従って、Azure Cosmos DB アカウントの資格情報を取得します。

Azure portal にサインインします。
Azure Cosmos DB アカウントに移動します。
[キー] ウィンドウを開き、アカウントの [URI] と [プライマリキー] をコピーします。

Azure Cosmos DB Emulator を使用している場合は、こちらの記事からエミュレーターの資格情報を取得してください。

手順 5: 一括実行のサポートを適用して CosmosClient オブジェクトを初期化する

生成された Program.cs ファイルをコードエディターで開きます。一括実行を有効にした CosmosClient の新しいインスタンスを作成し、それを使用して Azure Cosmos DB に対して操作を実行します。

まず、既定の Main メソッドを上書きし、グローバル変数を定義します。これらのグローバル変数には、エンドポイントと認可キー、データベースの名前、作成するコンテナー、一括で挿入する項目の数が含まれます。環境に応じて、endpointURL と承認キーの値を必ず置き換えてください。

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Text.Json;
using System.Threading.Tasks;
using Microsoft.Azure.Cosmos;

public class Program
{
     private const string EndpointUrl = "https://<your-account>.documents.azure.com:443/";
     private const string AuthorizationKey = "<your-account-key>";
     private const string DatabaseName = "bulk-tutorial";
     private const string ContainerName = "items";
     private const int AmountToInsert = 300000;

     static async Task Main(string[] args)
     {

     }
}

Main メソッド内に、CosmosClient オブジェクトを初期化する次のコードを追加します。

CosmosClient cosmosClient = new CosmosClient(EndpointUrl, AuthorizationKey, new CosmosClientOptions() { AllowBulkExecution = true });

注意

CosmosClientOptions で一括実行を指定すると、CosmosClient の有効期間中は実質的に変更できません。値を変更しても効果は得られません。

一括実行を有効にすると、CosmosClient によって同時実行操作が単一のサービス呼び出しに内部的にグループ化されます。これにより、サービス呼び出しを複数のパーティションに分散し、最終的に個々の結果を元の呼び出し元に割り当てることで、スループット利用率が最適化されます。

その後、すべての項目を格納するコンテナーを作成できます。パーティションキーとして /pk、プロビジョニングスループットとして 50000 RU/秒、すべてのフィールドを除外して書き込みスループットを最適化するカスタムインデックス作成ポリシーを定義します。 CosmosClient 初期化ステートメントの後に次のコードを追加します。

Database database = await cosmosClient.CreateDatabaseIfNotExistsAsync(Program.DatabaseName);

await database.DefineContainer(Program.ContainerName, "/pk")
        .WithIndexingPolicy()
            .WithIndexingMode(IndexingMode.Consistent)
            .WithIncludedPaths()
                .Attach()
            .WithExcludedPaths()
                .Path("/*")
                .Attach()
        .Attach()
    .CreateAsync(50000);

手順 6: 同時実行タスクのリストを設定する

一括実行のサポートを利用するには、データのソースと実行する操作に基づいて非同期タスクのリストを作成し、Task.WhenAll を使用してそれらを同時に実行します。まず、"Bogus" データを使用して、データモデルの項目のリストを生成します。実際のアプリケーションでは、項目は目的のデータソースから取得されます。

まず、dotnet add package コマンドを使用して、Bogus パッケージをソリューションに追加します。

dotnet add package Bogus

保存する項目を定義します。 Program.cs ファイル内に Item クラスを定義する必要があります。

public class Item
{
    public string id {get;set;}
    public string pk {get;set;}

    public string username{get;set;}
}

次に、Program クラス内にヘルパー関数を作成します。このヘルパー関数は、挿入対象として定義された項目の数を取得し、ランダムデータを生成します。

private static IReadOnlyCollection<Item> GetItemsToInsert()
{
    return new Bogus.Faker<Item>()
    .StrictMode(true)
    //Generate item
    .RuleFor(o => o.id, f => Guid.NewGuid().ToString()) //id
    .RuleFor(o => o.username, f => f.Internet.UserName())
    .RuleFor(o => o.pk, (f, o) => o.id) //partitionkey
    .Generate(AmountToInsert);
}

ヘルパー関数を使用して、操作するドキュメントのリストを初期化します。

IReadOnlyCollection<Item> itemsToInsert = Program.GetItemsToInsert();

次に、ドキュメントのリストを使用して同時実行タスクを作成し、タスクリストを設定して項目をコンテナーに挿入します。この操作を実行するには、Program クラスに次のコードを追加します。

Container container = database.GetContainer(ContainerName);
List<Task> tasks = new List<Task>(AmountToInsert);
foreach (Item item in itemsToInsert)
{
    tasks.Add(container.CreateItemAsync(item, new PartitionKey(item.pk))
        .ContinueWith(itemResponse =>
        {
            if (!itemResponse.IsCompletedSuccessfully)
            {
                AggregateException innerExceptions = itemResponse.Exception.Flatten();
                if (innerExceptions.InnerExceptions.FirstOrDefault(innerEx => innerEx is CosmosException) is CosmosException cosmosException)
                {
                    Console.WriteLine($"Received {cosmosException.StatusCode} ({cosmosException.Message}).");
                }
                else
                {
                    Console.WriteLine($"Exception {innerExceptions.InnerExceptions.FirstOrDefault()}.");
                }
            }
        }));
}

// Wait until all are done
await Task.WhenAll(tasks);

概要セクションで説明したように、これらの同時ポイント操作はすべてまとめて (つまり一括で) 実行されます。

手順 7: サンプルを実行する

サンプルは、dotnet コマンドで簡単に実行できます。

dotnet run

完全なサンプルを入手する

このチュートリアルの手順を実行する時間がない場合や、コードサンプルをダウンロードするだけの場合は、GitHub から入手できます。

プロジェクトをクローンしたら、Program.cs 内の必要な資格情報を必ず更新してください。

サンプルを実行するには、リポジトリディレクトリに変更し、dotnet を使用します。

cd cosmos-dotnet-bulk-import-throughput-optimizer
dotnet run

次のステップ

このチュートリアルでは、次の手順を実行しました。

Azure Cosmos DB アカウントを作成する
プロジェクトを構成する
一括サポートが有効になっている Azure Cosmos DB アカウントに接続する
同時作成操作によるデータインポートを実行する

これで、次のチュートリアルに進むことができます。

NoSQL 用 API を使って Azure Cosmos DB に対するクエリを実行する

Azure Cosmos DB への移行のための容量計画を実行しようとしていますか? 容量計画のために、既存のデータベースクラスターに関する情報を使用できます。

既存のデータベースクラスター内の仮想コアとサーバーの数のみがわかっている場合は、仮想コアまたは vCPU を使用した要求ユニットの見積もりに関するページを参照してください
現在のデータベースワークロードに対する通常の要求レートがわかっている場合は、Azure Cosmos DB Capacity Planner を使用した要求ユニットの見積もりに関するページを参照してください