Массовый импорт данных в учетную запись Azure Cosmos DB для NoSQL с помощью пакета SDK для .NET

  • Статья

ОБЛАСТЬ ПРИМЕНЕНИЯ: NoSQL

В этом руководстве описывается, как создать консольное приложение .NET, которое оптимизирует подготовленную пропускную способность (ЕЗ/с), необходимую для импорта данных в Azure Cosmos DB.

В этой статье вы прочитаете данные из примера источника данных и импортируете их в контейнер Azure Cosmos DB. Для работы с этим руководством используется версия 3.0 или более поздняя версия пакета SDK .NET для Azure Cosmos DB, предназначенного для .NET Framework или .NET Core.

Темы, рассматриваемые в этом руководстве:

  • Создание учетной записи Azure Cosmos DB
  • Настройка проекта.
  • Подключение к учетной записи Azure Cosmos DB с включенной массовой поддержкой
  • Импорт данных с помощью параллельных операций создания.

Предварительные требования

Перед выполнением инструкций, приведенных в этой статье, подготовьте следующие ресурсы:

Шаг 1. создание учетной записи Azure Cosmos DB;

Создайте учетную запись Azure Cosmos DB для NoSQL из портал Azure или с помощью эмулятора Azure Cosmos DB.

Шаг 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

Оставаясь в каталоге приложения, установите клиентскую библиотеку Azure Cosmos DB для .NET Core с помощью команды.

dotnet add package Microsoft.Azure.Cosmos

Шаг 4. Получение учетных данных учетной записи Azure Cosmos DB

Пример приложения должен пройти проверку подлинности в учетной записи Azure Cosmos DB. Для проверки подлинности необходимо передать в приложение учетные данные учетной записи Azure Cosmos DB. Получите учетные данные учетной записи Azure Cosmos DB, выполнив следующие действия.

  1. Войдите на портал Azure.
  2. Перейдите к своей учетной записи Azure Cosmos DB.
  3. Откройте панель Ключи и скопируйте URI и первичный ключ своей учетной записи.

Если вы используете эмулятор Azure Cosmos DB, получите учетные данные эмулятора из этой статьи.

Шаг 5. Инициализация объекта CosmosClient с поддержкой массового выполнения

Откройте созданный файл Program.cs в редакторе кода. Вы создадите экземпляр CosmosClient с включенным массовым выполнением и будете использовать его для выполнения операций с Azure Cosmos DB.

Начнем с перезаписи метода по умолчанию Main и определения глобальных переменных. Эти глобальные переменные включают в себя конечную точку и ключи авторизации, имя базы данных, контейнер, который вы создадите, и ряд элементов, которые вы вставите в массовом режиме. Обязательно замените значения URL-адреса конечной точки и ключей авторизации в соответствии со своей средой.

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 в качестве ключа секции, подготовленную пропускную способность в 50 000 ЕЗ/с и настраиваемую политику индексации, которая будет исключать все поля для оптимизации пропускной способности записи. Добавьте приведенный ниже код после инструкции инициализации 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.

dotnet add package Bogus

Добавьте определение элементов, которые необходимо сохранить. Необходимо определить класс Item в файле Program.cs.

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 с включенной массовой поддержкой
  • Импорт данных с помощью параллельных операций создания.

Теперь вы можете перейти к следующему руководству:

Запрос Azure Cosmos DB с помощью API для NoSQL

Пытаетесь выполнить планирование ресурсов для миграции в Azure Cosmos DB? Можете использовать для этого сведения о существующем кластере базы данных.