从 .NET Core 将 JSON 文件批量上传/导入到 Azure Cosmos DB

Question

我正在

JSON

应用程序中生成一些

.NET Core

文件。我想在创建

json

文件后立即将其导入到

Azure Cosmos DB

。

有什么方法可以通过

.NET Core

代码实现它吗？

Answer 1

根据我的测试，如果要在.Net core应用程序中使用Cosmos DB Bulk insert，则需要使用.Net CosmosDB SDK V3，并且其版本必须大于3.4.0。更多详情请参阅文档。

我的

.json

文件

[{
        "id": "1",
        "name": "test1",
        "age": "20"
    }, {
        "id": "2",
        "name": "test2",
        "age": "21"
    }, {
        "id": "3",
        "name": "test3",
        "age": "22"
    }, {
        "id": "4",
        "name": "test4",
        "age": "23"
    },
    {
        "id": "5",
        "name": "test5",
        "age": "24"
    }, {
        "id": "6",
        "name": "test6",
        "age": "25"
    }, {
        "id": "7",
        "name": "test7",
        "age": "26"
    }, {
        "id": "8",
        "name": "test8",
        "age": "27"
    }
]

我的代码

 private const string EndpointUrl = "";
        private const string AuthorizationKey = "";
        private const string DatabaseName = "testbulk";
        private const string ContainerName = "items";
        async static Task Main(string[] args)
        {
            string json = File.ReadAllText(@"E:\test.json");
            
            List<Item> lists = JsonConvert.DeserializeObject<List<Item>>(json);

            CosmosClientOptions options = new CosmosClientOptions() { AllowBulkExecution = true };
            CosmosClient cosmosClient = new CosmosClient(EndpointUrl, AuthorizationKey,options);
            
            Database database = await cosmosClient.CreateDatabaseIfNotExistsAsync(DatabaseName);
            Console.WriteLine(database.Id);
            Container container = await database.CreateContainerIfNotExistsAsync(ContainerName, "/id");
            Console.WriteLine(container.Id);
            
            List<Task> tasks = new List<Task>();
            foreach (Item item in lists)
            {
                tasks.Add(container.CreateItemAsync(item, new PartitionKey(item.Id))
                    .ContinueWith((Task<ItemResponse<Item>> task) =>
                    {


                        Console.WriteLine("Status: " + task.Result.StatusCode + "    Resource: "+ task.Result.Resource.Id );



                    }));


            }
            await Task.WhenAll(tasks);
}
class Item {

        [JsonProperty(PropertyName = "id")]
        public string Id { get; set; }

        [JsonProperty(PropertyName = "name")]
        public string Name { get; set; }

        [JsonProperty(PropertyName = "age")]
        public string Age { get; set; }
    }

有关如何开发应用程序的更多详细信息，请参阅博客

---

更新

我在 .Net 4.6.1 控制台应用程序中运行我的代码。

我的代码

 class Program
    {

        private const string EndpointUrl = "";
        private const string AuthorizationKey = "";
        private const string DatabaseName = "testbulk";
        private const string ContainerName = "items";
        async static Task Main(string[] args)
        {
            string json = File.ReadAllText(@"E:\test.json");

            List<Item> lists = JsonConvert.DeserializeObject<List<Item>>(json);

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

            Database database = await cosmosClient.CreateDatabaseIfNotExistsAsync(DatabaseName);
            Console.WriteLine(database.Id);
            Container container = await database.CreateContainerIfNotExistsAsync(ContainerName, "/id");
            Console.WriteLine(container.Id);

            List<Task> tasks = new List<Task>();
            foreach (Item item in lists)
            {
                tasks.Add(container.CreateItemAsync(item, new PartitionKey(item.Id))
                    .ContinueWith((Task<ItemResponse<Item>> task) =>
                    {


                        Console.WriteLine("Status: " + task.Result.StatusCode + "    Resource: " + task.Result.Resource.Id);



                    }));


            }
            await Task.WhenAll(tasks);
            Console.ReadLine();
        }
        class Item
        {

            [JsonProperty(PropertyName = "id")]
            public string Id { get; set; }

            [JsonProperty(PropertyName = "name")]
            public string Name { get; set; }

            [JsonProperty(PropertyName = "age")]
            public string Age { get; set; }
        }
    }

我的

package.json

<?xml version="1.0" encoding="utf-8"?>
<packages>
  <package id="Microsoft.Azure.Cosmos" version="3.4.1" targetFramework="net461" />
  <package id="Newtonsoft.Json" version="10.0.2" targetFramework="net461" />
  <package id="System.Buffers" version="4.4.0" targetFramework="net461" />
  <package id="System.Configuration.ConfigurationManager" version="4.5.0" targetFramework="net461" />
  <package id="System.Memory" version="4.5.1" targetFramework="net461" />
  <package id="System.Numerics.Vectors" version="4.4.0" targetFramework="net461" />
  <package id="System.Runtime.CompilerServices.Unsafe" version="4.5.1" targetFramework="net461" />
  <package id="System.Security.AccessControl" version="4.5.0" targetFramework="net461" />
  <package id="System.Security.Permissions" version="4.5.0" targetFramework="net461" />
  <package id="System.Security.Principal.Windows" version="4.5.0" targetFramework="net461" />
  <package id="System.ServiceModel.Primitives" version="4.5.0" targetFramework="net461" />
  <package id="System.Threading.Tasks.Extensions" version="4.5.1" targetFramework="net461" />
  <package id="System.ValueTuple" version="4.5.0" targetFramework="net461" />
</packages>

Answer 2

以下是如何在架构上处理此场景，该场景将是可扩展的并且符合指南：

• 从.Net Core应用程序中，当您创建json文件时，将其存储在Azure blob存储中
• 下一步，您将使用 Azure Blob 存储触发器创建一个 Azure Function，您可以查看此链接以供参考：

https://learn.microsoft.com/en-us/azure/azure-functions/functions-create-storage-blob-triggered-function

• 从 azure 函数中，您可以配置使用 BulkImport .net 库来更新 cosmos db 中的记录。

    BulkImportResponse bulkImportResponse = await bulkExecutor.BulkImportAsync(
      documents: documentsToImportInBatch,
      enableUpsert: true,
      disableAutomaticIdGeneration: true,
      maxConcurrencyPerPartitionKeyRange: null,
      maxInMemorySortingBatchSize: null,
      cancellationToken: token);

此 .NET 库提供了批量导入 API 的两种重载 - 一种接受序列化 JSON 文档列表，另一种接受反序列化 POCO 文档列表。要了解有关每个重载方法的定义的更多信息，请参阅 API 文档。

查看此链接以获取更多参考：

https://learn.microsoft.com/en-us/azure/cosmos-db/bulk-executor-dot-net

以上软件设计满足可用性、可扩展性和可维护性的期望。

欢迎在您的对话中标记我，希望对您有所帮助。