Início / coding / Perguntas / 79554143
Accepted
user989988
Asked: 2025-04-04 06:38:05 +0800 CST

anexar blob com .NET anexando dados a matrizes separadas

  • 772

Tenho uma função SubOrchestrator que chama as seguintes Funções de Atividade:

    public async Task GetTransitiveMembers(IDurableOrchestrationContext context, GroupMembershipRequest request)
    {
        if (!context.IsReplaying) _ = _log.LogMessageAsync(new LogMessage { RunId = request.RunId, Message = $"Getting results from 1st page" });
        var nextPageUrl = await context.CallActivityAsync<string>(nameof(MembersReaderFunction), new MembersReaderRequest { RunId = request.RunId, GroupId = request.SourceGroup.ObjectId, TargetGroupId = request.GroupId, CurrentPart = request.CurrentPart });
    
    
        while (!string.IsNullOrEmpty(nextPageUrl))
        {
            if (!context.IsReplaying) _ = _log.LogMessageAsync(new LogMessage { RunId = request.RunId, Message = $"Getting results from next page url: {nextPageUrl}" });
            var oldNextPageUrl = nextPageUrl;              
            nextPageUrl = await context.CallActivityAsync<string>(nameof(SubsequentMembersReaderFunction), new SubsequentMembersReaderRequest { RunId = request.RunId, NextPageUrl = nextPageUrl, GroupId = request.SourceGroup.ObjectId, TargetGroupId = request.GroupId, CurrentPart = request.CurrentPart })
        }
    }

Funções de atividade:

    [FunctionName(nameof(MembersReaderFunction))]
    public async Task<string> GetMembersAsync([ActivityTrigger] MembersReaderRequest request)
    {
        await _log.LogMessageAsync(new LogMessage { Message = $"{nameof(MembersReaderFunction)} function started", RunId = request.RunId }, VerbosityLevel.DEBUG);
        _calculator.RunId = request.RunId;
        var response = await _calculator.GetFirstTransitiveMembersPageAsync(request.GroupId, request.RunId);
        response.Users = new List<AzureADUser> { new AzureADUser { UserPrincipalName = "a" } };
        var fileName = $"/{request.TargetGroupId}/{request.RunId}_GroupMembership_{request.CurrentPart}.json";
        using (MemoryStream logEntryStream = new MemoryStream())
        {
            await JsonSerializer.SerializeAsync(logEntryStream, response.Users);
            logEntryStream.Position = 0;
            await _blobStorageRepository.AppendDataToBlobAsync(logEntryStream, fileName);
        }
        await _log.LogMessageAsync(new LogMessage { Message = $"{nameof(MembersReaderFunction)} function completed", RunId = request.RunId }, VerbosityLevel.DEBUG);
        return response.NextPageUrl;
    }

    [FunctionName(nameof(SubsequentMembersReaderFunction))]
    public async Task<string> GetMembersAsync([ActivityTrigger] SubsequentMembersReaderRequest request)
    {
        await _log.LogMessageAsync(new LogMessage { Message = $"{nameof(SubsequentMembersReaderFunction)} function started", RunId = request.RunId }, VerbosityLevel.DEBUG);
        _calculator.RunId = request.RunId;
        var response = await _calculator.GetNextTransitiveMembersPageAsync(request.NextPageUrl);
        response.Users = new List<AzureADUser> { new AzureADUser { UserPrincipalName = "b" } };
        var fileName = $"/{request.TargetGroupId}/{request.RunId}_GroupMembership_{request.CurrentPart}.json";
        using (MemoryStream logEntryStream = new MemoryStream())
        {
            await JsonSerializer.SerializeAsync(logEntryStream, response.Users);
            logEntryStream.Position = 0;
            await _blobStorageRepository.AppendDataToBlobAsync(logEntryStream, fileName);
        }
        await _log.LogMessageAsync(new LogMessage { Message = $"{nameof(SubsequentMembersReaderFunction)} function completed", RunId = request.RunId }, VerbosityLevel.DEBUG);
        return response.NextPageUrl;
    }

Repositório BlobStorage:

    public async Task AppendDataToBlobAsync (MemoryStream logEntryStream, string logBlobName)
    {
        var appendBlobClient = _containerClient.GetAppendBlobClient(logBlobName);
    
        await appendBlobClient.CreateIfNotExistsAsync();
        logEntryStream.Position = 0;
    
        var maxBlockSize = appendBlobClient.AppendBlobMaxAppendBlockBytes;
        var bytesLeft = logEntryStream.Length;
        var buffer = new byte[maxBlockSize];
        while (bytesLeft > 0)
        {
            var blockSize = (int)Math.Min(bytesLeft, maxBlockSize);
            var bytesRead = await logEntryStream.ReadAsync(buffer.AsMemory(0, blockSize));
            await using (MemoryStream memoryStream = new MemoryStream(buffer, 0, bytesRead))
            {
                await appendBlobClient.AppendBlockAsync(memoryStream);
            }
            bytesLeft -= bytesRead;
        }
    }

Eu vejo o conteúdo no blob assim:

    [
        {
            "ObjectId": "00000000-0000-0000-0000-000000000000",
            "Mail": null,
            "UserPrincipalName": "a",
            "Properties": null,
            "MembershipAction": null,
            "SourceGroup": "00000000-0000-0000-0000-000000000000",
            "SourceGroups": null
        }
    ][
        {
            "ObjectId": "00000000-0000-0000-0000-000000000000",
            "Mail": null,
            "UserPrincipalName": "b",
            "Properties": null,
            "MembershipAction": null,
            "SourceGroup": "00000000-0000-0000-0000-000000000000",
            "SourceGroups": null
        }
    ][
        {
            "ObjectId": "00000000-0000-0000-0000-000000000000",
            "Mail": null,
            "UserPrincipalName": "b",
            "Properties": null,
            "MembershipAction": null,
            "SourceGroup": "00000000-0000-0000-0000-000000000000",
            "SourceGroups": null
        }
    ]

Deveria ser:

    [
        {
            "ObjectId": "00000000-0000-0000-0000-000000000000",
            "Mail": null,
            "UserPrincipalName": "a",
            "Properties": null,
            "MembershipAction": null,
            "SourceGroup": "00000000-0000-0000-0000-000000000000",
            "SourceGroups": null
        },
        {
            "ObjectId": "00000000-0000-0000-0000-000000000000",
            "Mail": null,
            "UserPrincipalName": "b",
            "Properties": null,
            "MembershipAction": null,
            "SourceGroup": "00000000-0000-0000-0000-000000000000",
            "SourceGroups": null
        },
        {
            "ObjectId": "00000000-0000-0000-0000-000000000000",
            "Mail": null,
            "UserPrincipalName": "b",
            "Properties": null,
            "MembershipAction": null,
            "SourceGroup": "00000000-0000-0000-0000-000000000000",
            "SourceGroups": null
        }
    ]

Como posso corrigir isso?

1 respostas

  1. Best Answer

    Por que a saída parece errada no momento

    Para começar, vamos explicar a diferença entre sua saída observada e o que você deseja.

    • O que você quer: uma única lista (ou seja, uma única matriz JSON) com muitos objetos nela.
    • Saída observada: uma lista de listas, em vez de uma lista.

    O motivo para isso é explicado por como você está escrevendo o JSON. As linhas relevantes do seu código são:

    // You create a list
response.Users = new List<AzureADUser> { new AzureADUser { UserPrincipalName = "b" } };

// ...

// Then you serialise the list and add it onto the end of the blob:
await JsonSerializer.SerializeAsync(logEntryStream, response.Users);
logEntryStream.Position = 0;
await _blobStorageRepository.AppendDataToBlobAsync(logEntryStream, fileName);

    Em outras palavras: no início da função, seu blob se parece com isto:

    [
    {
        // original object
    }
]

    ...que é uma lista com 1 objeto.

    Mas seu código atual apenas serializa outra lista e tenta colá-la no final do JSON. Ele não está realmente adicionando-os à lista existente. Então você acaba com:

    [
    {
        // original object
    }
]
[
    {
        // new object
    }
]

    Não é mais nem um JSON válido.

    Por que não é prático fazer exatamente o que você está pedindo

    Se você pensar no que é necessário, você precisaria de alguma forma manipular o JSON existente, para que seus novos dados sejam inseridos naquele array. Como:

    [
    {
        // original object        
    }
    , // Now insert an extra comma here
    // Now insert your new objects here
    {

    }
] // And make sure you keep the closing bracket at the end.

    Isso é, realmente, quase impossível de fazer. Raspar bytes do final do blob, adicionar mais dados, colocar o colchete de volta no final... Bem impraticável.

    Você não pode adicionar dados sensatamente em um array JSON existente, a menos que você primeiro leia o arquivo inteiro e o desserialize em sua lista. Isso nos leva às soluções potenciais.

    Solução 1: desserialize todo o arquivo JSON em uma lista primeiro

    Foi isso que eu sugeri nos comentários. Você leria tudo, então adicionaria seus novos itens ao final da lista e escreveria de volta.

    Mas eu entendo perfeitamente que você não queira fazer isso, porque há uma grande queda de desempenho se você estiver lendo e escrevendo 300.000 itens em cada lista.

    Solução 2: armazene cada lote de itens em seu próprio arquivo blob e junte-os mais tarde

    Esta é uma solução muito mais escalável.

    Para sua função aqui:

        [FunctionName(nameof(SubsequentMembersReaderFunction))]
    public async Task<string> GetMembersAsync([ActivityTrigger] SubsequentMembersReaderRequest request)

    ...faça com que ele escreva um arquivo JSON novinho em folha com um nome diferente. Você pode então escrever esses 1.000 novos itens muito rapidamente; enquanto deixa os anteriores intactos em seus outros arquivos.

    Imagino que você dirá que isso causará problemas quando ler os arquivos JSON mais tarde (para o que quer que o resto do seu aplicativo faça). Você pode pensar que será muito lento/inconveniente ler 300 arquivos JSON de 1.000 itens; comparado a ler 1 arquivo JSON de 300.000 itens.

    Então, deixe-me convencê-lo de que é fácil ler muitos blobs. Você pode até mesmo paralelizá-los, o que eu mostrei no exemplo. (note que paralelizar não vale a pena se cada arquivo for muito grande; mas se você acabar com muitos arquivos pequenos, isso pode reduzir a latência.)

    Recuperar os arquivos em paralelo pode ser algo assim:

    // Get a list of blobs matching a prefix.
// These would be your 300 JSON files that each have a batch of 1,000 items.        
var blobs = container.GetBlobsAsync(prefix: "blobfile-");
var blobReaderTasks = new List<Task<Stream>>();

// Enumerate the list of blobs.
// This doesn't actually read them - it just gets the streams ready.
await foreach (BlobItem blobItem in blobs)
{
    blobReaderTasks.Add(container.GetBlobClient(blobItem.Name).OpenReadAsync());
}
var blobStreams = await Task.WhenAll(blobReaderTasks);

// For each stream, create a lambda function that will deserialise them into your object.
// Again, this doesn't actually parse the data yet - it just gets a bunch of tasks ready.
var blobDeserialisationTasks = blobStreams
    .Select(async stream => await JsonSerializer.DeserializeAsync<AzureADUser>(stream) ?? throw new Exception("Was null"))
    .ToArray();

// Now do the actual reading of the streams and deserialising,
// which will be done with a degree of parallelisation if you want that.
var adUsers = await Task.WhenAll(blobDeserialisationTasks);

    Espero que ajude.

    • 1

relate perguntas