Как многопоточность больших данных в 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 записей), а Orchestrator не продолжает работу, так как CallActivityAsyn c не может вернуть более 8 МБ данных.

Это также может привести к сбою, потому что он по существу попытается запустить 60000 операций для каждой записи.

Я делаю это так, Azure будет выполнять запись в ADL с использованием нескольких потоков. , Сначала я попробовал использовать семафоры, и несколько источников в Интернете сказали мне, что вместо этого следует использовать семпахоры, а не «CallActivityAsyn c», что позволит Azure управлять своими собственными потоками.

Как мне решить это и добиться многопоточной записи в ADL?

Для записи я использую библиотеку, которая может записывать только один файл за раз (я знаю, что новая библиотека от MS включает в себя Bulk Write, но я я не могу использовать это по разным причинам).

Answer 1

Есть ли причина для GetRecords и WriteRecord в настройке функции Durable? Если нет, то GetRecords может отбросить каждый Record объект (сериализованный до JSON) в Azure Queue / EventHub вместо возврата огромного списка. Затем WriteRecords может быть запущен из этой очереди / концентратора событий для обработки каждого сообщения.