如何在 Orchestrator 中对大数据进行多线程处理？

Question

我在 Orchestrator 中有以下代码：

        var parallelTasks = new List<Task>();

        // Get Records
        List<Record> records = await context.CallActivityAsync<List<Record>>("GetRecords", orchestrationContext);

        // Write Records
        foreach (Record record in records)
        {
            parallelTasks.Add(context.CallActivityAsync<int>("WriteRecord", record));
        }

        await Task.WhenAll(parallelTasks);

这失败了，因为 GetRecords 返回了太多数据（60000 条记录），并且由于 CallActivityAsync 无法返回超过 8mb 的数据，Orchestrator 无法继续。

这也可能会失败，因为它实际上会尝试为每次写入启动 60000 个活动。

我这样做是为了让 Azure 使用多个线程写入 ADL。一开始我尝试使用信号量，网上的多个消息来源告诉我，不应使用信号量，而应使用“CallActivityAsync”，这将允许 Azure 管理自己的线程。

如何解决这个问题并实现多线程写入ADL？

为了记录，我使用的库一次只能写入一个文件（我知道 MS 的新库包含批量写入，但由于不同的原因我无法使用它）。

Answer 1

GetRecords 和 WriteRecord 是否有理由使用 Durable 功能设置？如果没有，GetRecords 可以将每个 Record 对象（序列化为 JSON）拖放到 Azure Queue/EventHub，而不是返回一个巨大的列表。然后可以从该队列/EventHub 触发WriteRecords 来处理每条消息。