c# - Azure CosmosDB : Improve Bulk insertion performance

Question

我正在使用 Azure Cosmos DB SDK(3.0) 进行 CRUD 操作。当我尝试同时插入 8,000-10,000 条记录时，大约需要 3-4 分钟。

这是我的代码:

  public async Task<ResultDto> HandleAsync(EnableOrDisableSubscriptionCommand command, ILogger logger)
        {
            logger.Info("Started EnableOrDisableSubscriptionCommand ", nameof(EnableOrDisableSubscriptionCommand));
            
            if (command.UiNotifications.Any())
            {
                await AddSubscription(command, SubscriptionAction.UiNotification, command.UiNotifications);
                logger.Info("Added UI notification subscriptions");
            }
            if (command.EmailNotifications.Any())
            {
                await AddSubscription(command, SubscriptionAction.Email, command.EmailNotifications);
                logger.Info("Added Email notification subscriptions");
            }

            return new ResultDto { ResultType = ResultType.Success, Message = $"User {command.UserId} SubscriptionStatus" };
        }

        
        private async Task AddSubscription(EnableOrDisableSubscriptionCommand command, SubscriptionAction subscriptionAction, IList<int> notificationCategoryTypes)
        {
            foreach (var notificationCategory in notificationCategoryTypes)
            {
                var notificationTypes = Utility.GetNotificationTypes((NotificationCategoryType)notificationCategory);

                foreach (var notificationType in notificationTypes)
                {
                    foreach (var payerAccountSubscriptions in command.Subscriptions)
                    {
                        if (payerAccountSubscriptions.AccountNumbers?.Any() ?? false)
                        {
                            foreach (var accountNumber in payerAccountSubscriptions.AccountNumbers.Where(a => !string.IsNullOrEmpty(a)))
                            {
                                await _repository.Create(subscriptionAction, notificationType,
                                     payerAccountSubscriptions.ColCoId, payerAccountSubscriptions.PayerNumber, accountNumber, command.UserRole,
                                     command.UserId);
                            }
                        }
                        else
                        {
                            await _repository.Create(subscriptionAction, notificationType,
                                payerAccountSubscriptions.ColCoId, payerAccountSubscriptions.PayerNumber, null, command.UserRole,
                                command.UserId);

                        }
                    }
                }
            }
        }

订阅存储库创建方法:

 public async Task Create(SubscriptionAction subscriptionAction, NotificationType notificationType,
            int colCoId, string payerNumber, string accountNumber, UserRole userRole, string userId, string cardId = null)
        {
            var eventType = Utility.GetEventType(notificationType);

            var subscriptionBase = new Subscription
            {
                Id = Guid.NewGuid(),
                IsActive = true,
                Action = subscriptionAction,
                ActionDesc = subscriptionAction.ToString(),
                Version = (int)SubscriptionVersion.V2,
                NotificationType = notificationType,
                NotificationTypeDesc = notificationType.ToString(),
                EventType = eventType,
                EventTypeDesc = eventType.ToString(),
                ColCoId = colCoId,
                PayerNumber = payerNumber,
                AccountNumber = accountNumber,
                CardId = cardId,
                DistributionGroups = new List<string> { userRole.ToString() },
                DistributionUserIds = new List<string> { userId }
            };
            await CreateItemAsync(subscriptionBase);
        }

通用存储库:

 public async Task<ItemResponse<T>> CreateItemAsync(T item)
        {
            return await _container.CreateItemAsync<T>(item);
        }

由于此问题，我的 Http 触发器 Azure Function 返回 System.OutOfMemoryException。

我该如何改进？

Answer 1

您可以通过在客户端中设置 AllowBulkExecution = true 并将每个插入操作添加到要执行的任务来改进这一点。

您可以了解更多信息并查看示例 here

编辑:(这太长了，无法添加为评论) 您需要多少 RU/秒取决于许多因素，包括您希望摄取数据的速度。我将测量插入其中一项的成本，然后将您的预配置吞吐量除以插入一项的量。结果应该是每秒可以插入的项目数(假设您没有执行其他操作)。如果您有一个项目的插入成本为 10 RU/s，并且您配置了 3000 RU/s，那么您每秒可以摄取 300 个项目。总共 10,000 个项目/每秒 300 个 = 33 秒。

因此，如果这需要 3-4 分钟，则说明您的代码存在其他问题。我会回去阅读我上面发布的文章，因为我没有看到您实现我们建议的模式，特别是将每个操作放在 List 对象上，然后调用 await Task.WhenAll(this.Tasks);

我看到的另一个问题是您没有在 InsertItemsAsync() 调用中指定分区键。这会将所有内容写入单个空分区，并且一旦大小达到 20GB，最终将停止接受任何新写入。