如何计算将批量导入任务导入 CosmosDB 所需的吞吐量？

Question

更新

我认为我的问题可能是客户问题。通过关闭 Visual Studio 2019 并重新打开它并运行，我可以完全加载一个文件（所有 200 万条记录）。但是，如果我再次尝试运行，它将失败并出现如下所述的问题。

Azure DocumentDB sporadically throws SocketException / GoneException

原始问题

我正在尝试在 cosmos DB 中批量加载一些数据文件。我正在使用我在此处找到的示例：

https://github.com/Azure/azure-cosmosdb-bulkexecutor-dotnet-getting-started/tree/master/BulkImportSample

在导入示例中，他们使用循环制作虚假文档。就我而言，我正在从目录中读取 CSV 文件。每个 CSV 文件行都将转换为一个 CosmosDB 文档。

默认吞吐量为 400，但我将其增加到 10000。导入后我不需要这么高，但我可以将其设置为批量导入任务所需的任何值。尽管如此，我还是遇到了某种吞吐量或节流问题，我对如何确定执行此数据导入所需的吞吐量感到困惑。这些 CSV 文件中的每一个都包含大约 200 万行，但每行只有 10 个标量值。

它开始工作了。我从 BulkExecutorTrace 获得了一些输出，例如“BulkExecutorTrace 信息：0：分区索引 0：2452 | 在 1 秒内以 19342.86 RU/s 的速度运行 2452 个文档，执行 4 个任务。面临 0 个限制”

但是在这样的 19 行输出之后，我得到：

DocDBTrace 警告：0：抛出异常：mscorlib.dll 中的“System.Threading.Tasks.TaskCanceledException” BulkExecutorTrace 信息：0：RNTBD 调用在通道 192.168.11.105:19676 -> 40.78.226.8:14319 上超时。错误：接收超时 分区索引 0 : 22068 |在 1 秒内以 9652.88 RU/s 的速度处理 1226 个文档，执行 20 个任务。面临0个油门 抛出异常：Microsoft.Azure.Documents.Client.dll 中的“Microsoft.Azure.Documents.TransportException” 抛出异常：mscorlib.dll 中的“Microsoft.Azure.Documents.TransportException” 抛出异常：mscorlib.dll 中的“Microsoft.Azure.Documents.TransportException” DocDBTrace 信息：0：RequestAsync 失败：RID：dbs/Diseases/colls/Diseases/sprocs/__.sys.commonBulkInsert，资源类型：StoredProcedure，Op：（操作类型：ExecuteJavaScript，资源类型：StoredProcedure），地址：rntbd://cdb -ms-prod-eastus1-fd10.documents.azure.com:14319/apps/00e9d5e0-018e-43a2-b5a4-f41c78498cdb/services/61a05d2a-fb30-455f-864e-c9e10e85684c/partitions/92daa841-dcc5-40 -2956cda4d2ac/replicas/132323601201339145p/，异常：Microsoft.Azure.Documents.TransportException：发生客户端传输错误：等待服务器响应时请求超时。 （时间：2020-04-26T18:18:32.9586759Z，活动 ID：80910ba7-8b36-40fa-a3bf-3eac239b00e2，错误代码：ReceiveTimeout [0x0010]，基本错误：HRESULT 0x80131500，URI：rntbd://cdb-ms -prod-eastus1-fd10.documents.azure.com:14319/apps/00e9d5e0-018e-43a2-b5a4-f41c78498cdb/services/61a05d2a-fb30-455f-864e-c9e10e85684c/partitions/92daa8241-dcc5-56cdaf04-cdaf02 /replicas/132323601201339145p/，连接：192.168.11.105:19676 -> 40.78.226.8:14319，发送的有效负载：True，CPU 历史记录：(2020-04-26T18:18:12.7679827Z 80.069)，(2020-04-2) 18:22.7667638Z 28.038), (2020-04-26T18:18:22.7672671Z 100.000), (2020-04-26T18:18:22.7672671Z 0.000), (2020-04-26T18:18:22.76726) 2020-04-26T18:18:32.7701961Z 20.629)，CPU 数量：8) 在 Microsoft.Azure.Documents.Rntbd.Channel.d__13.MoveNext() --- 从先前抛出异常的位置结束堆栈跟踪 --- 在 System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess（任务任务） 在 System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification（任务任务） 在 Microsoft.Azure.Documents.Rntbd.LoadBalancingPartition.d__9.MoveNext()

private async Task RunBulkImportAsync()
    {
        DocumentCollection dataCollection = null;

        try
        {
            dataCollection = GetCollectionIfExists(client, DatabaseName, CollectionName);
            if (dataCollection == null)
            {
                throw new Exception("The data collection does not exist");
            }
        }
        catch (Exception de)
        {
            Trace.TraceError("Unable to initialize, exception message: {0}", de.Message);
            throw;
        }

        string partitionKeyProperty = dataCollection.PartitionKey.Paths[0].Replace("/", "");

        // Set retry options high for initialization (default values).
        client.ConnectionPolicy.RetryOptions.MaxRetryWaitTimeInSeconds = 30;
        client.ConnectionPolicy.RetryOptions.MaxRetryAttemptsOnThrottledRequests = 9;

        IBulkExecutor bulkExecutor = new BulkExecutor(client, dataCollection);
        await bulkExecutor.InitializeAsync();

        // Set retries to 0 to pass control to bulk executor.
        client.ConnectionPolicy.RetryOptions.MaxRetryWaitTimeInSeconds = 0;
        client.ConnectionPolicy.RetryOptions.MaxRetryAttemptsOnThrottledRequests = 0;

        BulkImportResponse bulkImportResponse = null;
        long totalNumberOfDocumentsInserted = 0;
        double totalRequestUnitsConsumed = 0;
        double totalTimeTakenSec = 0;

        var tokenSource = new CancellationTokenSource();
        var token = tokenSource.Token;

        foreach (string d in Directory.GetDirectories(RootPath).Take(1))
        {

            foreach (string f in Directory.GetFiles(d).Take(1))
            {

                Trace.WriteLine("Processing file, " + f);

                var lines = File.ReadAllLines(f);

                Trace.WriteLine("File has " + lines.Count() + "lines");

                List<RowToImport> dataToImport = lines
                                       .Skip(1)
                                       .Select(v => RowToImport.FromCsv(v))
                                       .ToList();

                List<string> documentsToImportInBatch = dataToImport.Select(dti => GenerateJsonDocument(Guid.NewGuid().ToString(), dti.Disease, dti.Year, dti.Age, dti.Country, dti.CountryName, dti.CohortSize, dti.DeathsCongenital)).ToList();

                // Invoke bulk import API.

                var tasks = new List<Task>();

                tasks.Add(Task.Run(async () =>
                {
                    Trace.TraceInformation(String.Format("Executing bulk import for batch {0}", f));

                    do
                    {
                        try
                        {
                            bulkImportResponse = await bulkExecutor.BulkImportAsync(
                                documents: documentsToImportInBatch,
                                enableUpsert: true,
                                disableAutomaticIdGeneration: true,
                                maxConcurrencyPerPartitionKeyRange: 100,
                                maxInMemorySortingBatchSize: null,
                                cancellationToken: token);
                        }
                        catch (DocumentClientException de)
                        {
                            Trace.TraceError("Document client exception: {0}", de);
                            //Console.WriteLine("Document client exception: {0} {1}", f, de);
                            break;
                        }
                        catch (Exception e)
                        {
                            Trace.TraceError("Exception: {0}", e);
                            //Console.WriteLine("Exception: {0} {1}", f, e);
                            break;
                        }
                    } while (bulkImportResponse.NumberOfDocumentsImported < documentsToImportInBatch.Count) ;


                    Trace.WriteLine(String.Format("\nSummary for batch {0}:", f));
                    Trace.WriteLine("--------------------------------------------------------------------- ");
                    Trace.WriteLine(String.Format("Inserted {0} docs @ {1} writes/s, {2} RU/s in {3} sec",
                        bulkImportResponse.NumberOfDocumentsImported,
                        Math.Round(bulkImportResponse.NumberOfDocumentsImported / bulkImportResponse.TotalTimeTaken.TotalSeconds),
                        Math.Round(bulkImportResponse.TotalRequestUnitsConsumed / bulkImportResponse.TotalTimeTaken.TotalSeconds),
                        bulkImportResponse.TotalTimeTaken.TotalSeconds));
                    Trace.WriteLine(String.Format("Average RU consumption per document: {0}",
                        (bulkImportResponse.TotalRequestUnitsConsumed / bulkImportResponse.NumberOfDocumentsImported)));
                    Trace.WriteLine("---------------------------------------------------------------------\n ");

                    totalNumberOfDocumentsInserted += bulkImportResponse.NumberOfDocumentsImported;
                    totalRequestUnitsConsumed += bulkImportResponse.TotalRequestUnitsConsumed;
                    totalTimeTakenSec += bulkImportResponse.TotalTimeTaken.TotalSeconds;
                },
                token));

                await Task.WhenAll(tasks);

            }

            Trace.WriteLine("Overall summary:");
            Trace.WriteLine("--------------------------------------------------------------------- ");
            Trace.WriteLine(String.Format("Inserted {0} docs @ {1} writes/s, {2} RU/s in {3} sec",
                totalNumberOfDocumentsInserted,
                Math.Round(totalNumberOfDocumentsInserted / totalTimeTakenSec),
                Math.Round(totalRequestUnitsConsumed / totalTimeTakenSec),
                totalTimeTakenSec));
            Trace.WriteLine(String.Format("Average RU consumption per document: {0}",
                (totalRequestUnitsConsumed / totalNumberOfDocumentsInserted)));
            Trace.WriteLine("--------------------------------------------------------------------- ");


            Trace.WriteLine("\nPress any key to exit.");
            Console.ReadKey();
        }

    }

在目录和文件循环中，我现在使用 Take(1) 只是为了尝试让一个文件工作。但是，实际上有 4 个目录，每个目录中有近百个文件。

你有什么建议可以告诉我我需要如何限制这个东西才能让它导入所有这些数据？

Answer 1

该异常与吞吐量无关。该异常指向超时/连接问题，SDK Troubleshooting page

中引用了该问题

在您的异常中，我们可以看到 CPU 出现峰值：

CPU history: 
(2020-04-26T18:18:12.7679827Z 80.069), 
(2020-04-26T18:18:22.7667638Z 28.038), 
(2020-04-26T18:18:22.7672671Z 100.000), 
(2020-04-26T18:18:22.7672671Z 0.000), 
(2020-04-26T18:18:22.7672671Z 0.000), 
(2020-04-26T18:18:32.7701961Z 20.629)

这可能会导致连接问题。如果您在本地开发机器上运行它，请查看其他哪些进程可能正在消耗 CPU。如果它在 VM 中运行，则可能需要更大的 CPU 池。

此外，根据您的代码，您正在使用来自并发操作的批量执行器（您正在并行创建多个任务）。

Bulk Executor performance tips 表示不应该这样做：

由于单个批量操作 API 执行会消耗大量客户端计算机的 CPU 和网络 IO（这是通过在内部生成多个任务来实现的）。避免在执行批量操作 API 调用的应用程序进程中产生多个并发任务。如果在单个虚拟机上运行的单个批量操作 API 调用无法消耗整个容器的吞吐量（如果您的容器的吞吐量 > 100 万 RU/s），则最好创建单独的虚拟机以并发执行批量操作 API来电。