Batch insert to Table Storage via Azure function

Question

I have a following azure storage queue trigger azure function which is binded to azure table for the output.

[FunctionName("TestFunction")]
public static async Task<IActionResult> Run(
    [QueueTrigger("myqueue", Connection = "connection")]string myQueueItem,
    [Table("TableXyzObject"), StorageAccount("connection")] IAsyncCollector<TableXyzObject> tableXyzObjectRecords)
{
            var tableAbcObject = new TableXyzObject();

            try
            {    
                tableAbcObject.PartitionKey = DateTime.UtcNow.ToString("MMddyyyy");
                tableAbcObject.RowKey = Guid.NewGuid();
                tableAbcObject.RandomString = myQueueItem;

                await tableXyzObjectRecords.AddAsync(tableAbcObject);
            }
            catch (Exception ex)
            {
            }
            
            return new OkObjectResult(tableAbcObject);
        }

        public class TableXyzObject : TableEntity
        {
            public string RandomString { get; set; }
        }
    }
}

I am looking for a way to read 15 messages from poisonqueue which is different than myqueue (queue trigger on above azure function) and batch insert it in to dynamic table (tableXyz, tableAbc etc) based on few conditions in the queue message. Since we have different poison queues, we want to pick up messages from multiple poison queues (name of the poison queue will be provided in the myqueue message). This is done to avoid to spinning up new azure function every time we have a new poison queue.

Following is the approach I have in my mind,
--> I might have to get 15 queue messages using queueClient (create new one) method - ReceiveMessages(15) of Azure.Storage.Queue package
--> And do a batch insert using TableBatchOperation class (cannot use output binding)

Is there any better approch than this?

You're using a QueueTrigger, but then want to call ReceiveMessages() inside of the queued trigger function back to the same queue? Azure Functions already has concurrency for storage queue triggers (default is 16 processes). Using ReceiveMessages() doesn't seem to make sense. Can you explain? If you cannot use output binding for whatever reason, then doing the inserts with a TableBatchOp makes sense. — Bryan Lewis
– Bryan Lewis, Commented Sep 22, 2020 at 23:32
sorry for the confusion. Queue (myqueue) trigger here is different than the queue which I want to read messages from. — Vicky
– Vicky, Commented Sep 23, 2020 at 2:25

PerfectlyPanda · Accepted Answer · 2020-09-23 16:18:48Z

2

Unfortunately, storage queues don't have a great solution for this. If you want it to be dynamic then the idea of implementing your own clients and table outputs is probably your best option. The one thing I would suggest changing is using a timer trigger instead of a queue trigger. If you are putting a message on your trigger queue every time you add something to the poison queue it would work as is, but if not a timer trigger ensures that poisoned messages are handled in a timely fashion.

Original Answer (incorrectly relating to Service Bus queues)

Bryan is correct that creating a new queue client inside your function isn't the best way to go about this. Fortunately, the Service Bus extension does allow batching. Unfortunately the docs haven't quite caught up yet.

Just make your trigger receive an array:

[QueueTrigger("myqueue", Connection = "connection")]string myQueueItem[]

You can set your max batch size in the host.json:

"extensions": {
  "serviceBus": {
    "batchOptions": {
      "maxMessageCount":  15
    }
  }
}

edited Sep 23, 2020 at 16:18

answered Sep 23, 2020 at 0:47

PerfectlyPanda

3,5111 gold badge9 silver badges18 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

Vicky Over a year ago

sorry for the confusion. Made an edit in the above question. -------> I am looking for a way to read 15 messages from poisonqueue which is different than myqueue (trigger on above function) and batch insert it in to dynamic table (tableXyz, tableAbc etc) based on few conditions in the queue message. Since we have different poison queues, we want to pick up messages from multiple poison queues (name of the poison queue will be provided in the myqueue message). This is done to avoid spinning up new azure function every time we have a new poison queue

PerfectlyPanda Over a year ago

I can't think of a way to make connections to dead letter queues dynamic-- it's a 1:1 relationship. However, you can create a seperate queue and use the "ForwardDeadLetteredMessagesTo" property to create a centralized dead letter queue for your function to read from. The setting isn't exposed in the portal so you'll need to use something like Service Bus Explorer or the Azure CLI to set it up.

Vicky Over a year ago

Thank you for quick response. Couple of questions, 1. does the Service Bus extension that allows batching supports Azure storage queues as well? I did not find proper document for the same. Also the string array dint work. 2. Also, is the "ForwardDeadLetteredMessagesTo" property available for azure storage queues which has poison queues?

PerfectlyPanda Over a year ago

That's 100% on me for missing the fact that you are using storage queues and not Service Bus (no idea how I missed it- you were very clear on that point). Service Bus queues have the property to allow automatic redirect and batching within Functions at an activity level. Storage queues have neither of those things. I've edited my answer so that the answer for the correct service (Storage Queues) isn't buried in the comments.

Collectives™ on Stack Overflow

Batch insert to Table Storage via Azure function

1 Answer 1

4 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Related