How to implement ForEachAsync?

Question

Questions on Lambda to delete partitions.

1. The existing query which uses parallelization is failing since it exceeds the number of parallel queries. We want to replace it with sequential queries and increased timeout for lambda.
2. Can we change the lambda to parallel with limited threads?

Database -> aws athena = Getting the List of clients from Athena. Looping throgh it.

Right now it works fine with sequential calls as well but since the number of clients is small now, it would pose a problem for future. The only issue with limited parallel threads is that we need some code to handle the thread count as well.

Then someone suggested me use this: https://devblogs.microsoft.com/pfxteam/implementing-a-simple-foreachasync-part-2/

https://gist.github.com/0xced/94f6c50d620e582e19913742dbd76eb6

public class AthenaClient {
    private readonly IAmazonAthena _client;
    private readonly string _databaseName;
    private readonly string _outputLocation;
    private readonly string _tableName;
    const int MaxQueryLength = 262144;
    readonly int _maxclientsToBeProcessed;
    
    public AthenaClient(string databaseName, string tableName, string outputLocation, int maxclientsToBeProcessed) {
        _databaseName = databaseName;
        _tableName = tableName;
        _outputLocation = outputLocation;
        _maxclientsToBeProcessed = maxclientsToBeProcessed == 0 ? 1 : maxclientsToBeProcessed;
        _client = new AmazonAthenaClient();
    }
    
    public async Task < bool > DeletePartitions() {
        var clients = await GetClients();
        for (int i = 0; i < clients.Count; i = i + _maxclientsToBeProcessed) {
            var clientItems = clients.Skip(i).Take(_maxclientsToBeProcessed);
            var queryBuilder = new StringBuilder();
            queryBuilder.AppendLine($ "ALTER TABLE { _databaseName }.{_tableName} DROP IF EXISTS");
            foreach(var client in clientItems) {
                queryBuilder.AppendLine($ " PARTITION (client_id = '{client}'), ");
            }
            var query = queryBuilder.ToString().Trim().TrimEnd(',') + ";";
            LambdaLogger.Log(query);
            if (query.Length >= MaxQueryLength) {
                throw new Exception("Delete partition query length exceeded.");
            }
            var queryExecutionId = StartQueryExecution(query).Result;
            await CheckQueryExecutionStatus(queryExecutionId);
        }
        return true;
    }
}

Answer 1

It seems that the actual question should be :

How can I change the database partitions for lots of clients in AWS Athena without executing them sequentially?

The answer isn't ForEachAsync or the upcoming await foreach in C# 8. An asynchronous loop would still send calls to the service one at a time, it "just" wouldn't block while waiting for an answer.

Concurrent workers

This is a concurrent worker problem that can be handled using eg the TPL Dataflow library's ActionBlock class or the new System.Threading.Channel classes.

The Dataflow library is meant to create event/message processing pipelines similar to a shell script pipeline, by moving data between independent blocks. Each block runs on its own task/thread which means you can get concurrent execution simply by breaking up processing into blocks.

It's also possible to increase the number of processing tasks per block, by specifying the MaxDegreeOfParallelism option when creating the block. This allows us to quickly create "workers" that can work on lots of messages concurrently.

Example

In this case, the "message" is the Client whatever that is. A single ActionBlock could create the DDL statement and execute it. Each block has an input queue which means we can just post messages to a block and await for it to execute everything using the DOP we specified.

We can also specify a limit to the queue so it won't get flooded if the worker tasks can't run fast enough :

var options=new ExecutionDataflowBlockOptions
     {
        MaxDegreeOfParallelism = _maxclientsToBeProcessed,
        BoundedCapacity = _maxclientsToBeProcessed*3, //Just a guess
     });
var block=new ActionBlock<Client>(client=>CreateAndRunDDL(client));

//Post the client requests
foreach(var client in clients)
{
    await block.SendAsync(client);
}

//Tell the block we're done
block.Complete();
//Await for all queued messages to finish processing
await block.Completion;

The CreateAndRunDDL(Client) method should do what the code inside the question's loop does. A good idea would be to refactor it though, and create separate functions to create and execute the query , eg :

async Task CreateAndRunDDL(Client client)
{
    var query = QueryForClient(...);
    LambdaLogger.Log(query);
    if (query.Length >= MaxQueryLength) {
        throw new Exception("Delete partition query length exceeded.");
    }
    var queryExecutionId = await StartQueryExecution(query);
    await CheckQueryExecutionStatus(queryExecutionId);
}

Blocks can be linked too. If we wanted to batch multiple clients together for processing, we can use a BatchBlock and feed its results to our action block, eg :

var batchClients = new BatchBlock<Client>(20);
var linkOptions = new DataflowLinkOptions
                  { 
                      PropagateCompletion = true
                  };

var block=new ActionBlock<Client>(clients=>CreateAndRunDDL(clients));
batchClients.LinkTo(block,linkOptions);

This time the CreateAndRunDDL method accepts a Client[] array with the number of clients/messages we specified in the batch size.

async Task CreateAndRunDDL(Client[] clients)
{
    var query = QueryForClients(clients);
    ...
}

Messages should be posted to the batchClients block now. Once that completes, we need to wait for the last block in the pipeline to complete :

foreach(var client in clients)
{
    await batchClients.SendAsync(client);
}

//Tell the *batch block* we're done
batchClient.Complete();
//Await for all queued messages to finish processing
await block.Completion;