Как лучше всего загрузить сериализованный IAsyncEnumerable в хранилище BLOB-объектов Azure?
Мы получаем множество записей с удаленного сервера в виде IAsyncEnumerable и после простого преобразования и сериализации JSON пытаемся загрузить результат (файл JSON) в Azure Blob Storage.
Могу ли я как-нибудь улучшить этот код, чтобы уменьшить максимальное использование памяти. Количество записей, которые мы получаем с удаленного сервера, велико.
public async Task ExecuteAsync()
{
var items = GetItems();
using var memoryStream = new MemoryStream();
await JsonSerializer.SerializeAsync(memoryStream, items);
var blobContainerClient = new Azure.Storage.Blobs.BlobContainerClient("connectionString", "container");
var blobClient = blobContainerClient.GetBlobClient("blobName");
memoryStream.Position = 0;
await blobClient.UploadAsync(memoryStream, overwrite: true);
}
private static async IAsyncEnumerable<Item> GetItems()
{
// This is calling a remote server that returns IAsyncEnumerable<Row>
await foreach (var row in GetIAsyncEnumerable())
{
yield return new Item(row);
}
}
До тех пор, пока с сериализацией JSON все не будет в порядке, мы получаем записи одну за другой, а преобразование типа объекта и сериализация также происходят одна за другой.
Но после await JsonSerializer.SerializeAsync(memoryStream, items) мы ждем, пока все элементы закончатся и материализуются в памяти, а затем начинаем загрузку.
Есть ли способ продолжить эту цепочку IAsyncEnumerable и начать загрузку, как только будет получен первый элемент, и продолжить загрузку для каждого нового объекта?
Ответы 1
Вы можете напрямую открыть поток для большого двоичного объекта и сериализовать в него элементы.
public async Task ExecuteAsync()
{
var blobContainerClient = new BlobContainerClient("connectionString", "container");
var blobClient = blobContainerClient.GetBlobClient("blobName");
var items = GetItems();
await using var blobStream = await blobClient.OpenWriteAsync(overwrite: true);
await JsonSerializer.SerializeAsync(blobStream, items);
}
Чтобы иметь возможность записывать элементы по мере их появления, вам понадобится что-то вроде этого:
public async Task ExecuteAsync()
{
var blobContainerClient = new BlobContainerClient("connectionString", "container");
var blobClient = blobContainerClient.GetBlobClient("blobName");
var items = GetItems();
await using var blobStream = await blobClient.OpenWriteAsync(overwrite: true);
await using var writer = new Utf8JsonWriter(blobStream, new JsonWriterOptions { Indented = true });
writer.WriteStartArray();
await foreach (var item in items)
{
JsonSerializer.Serialize(writer, item);
await writer.FlushAsync(); // If you want the item immediately to be available in the blob. For better performance remove.
}
writer.WriteEndArray();
await writer.FlushAsync();
}
@Магнус, спасибо за ответ. Я изменил SerializeAsync на Serialize во втором примере, чтобы все заработало. Кажется, это очень хороший подход, но результат очень медленный, примерно на два-три порядка медленнее, чем оригинал, и я не знаю, почему.
Попробуйте убрать флеш в цикле. Затем он будет сброшен, когда буфер заполнится.
Неужели нужно
FlushAsync()после каждого предмета?