Azure durable function for bulk insert toward Postgres DB

1
votes

my Azure's serverless architecture is composed from some producers, a service bus, un azure function and a Postgres DB (see the below picture).

The Postgres DB is a legacy requirement, I cannot change it.

The following is the operations flow:

  1. The producers send a message with a temporal frequency towards service bus (about 9000 messagges/minute). Every producers sends a single message.
  2. The azure functions consumes the message and insert a row into the Postgres DB

In order to avoid a strong load of DB and open a lot of connections I would to aggregate the messages into the function and inset them by bulk insert. Could I work well with a durable function (entity function)?

Can you help me please? Best Regards

enter image description here

2
postgresqlazureazure-functionsazureservicebusazure-durable-functions
Each function invocation should receive a batch of messages anyway already and not being called per single message github.com/Azure/azure-functions-servicebus-extension/issues/… – silent
Great, can I configure it? – Michel Foucault
what do you need to configure? – silent
Looking at the PR, I would say: yes. but not sure how exactly. Might be worth to open a issue in the github repo. default seems to be 1000: github.com/Azure/azure-functions-servicebus-extension/pull/39/… – silent
got it to work. see my answer below – silent

2 Answers

1
votes

I would consider Durable Functions just in case if you have multiple steps, so then you can build your workflow using Durable Functions. In your case, what I would do is to play around with the maxConcurrentCalls parameter on your hosts.json file, so you can then set an acceptable value for concurrent calls for your DB. Please reference this sample to get more info.

1
votes

Ok, got it to work as discussed in the comments. You don't need a durable function but you should receive (and then write) messages in batches from the Service Bus. Here is a batching example:

[FunctionName("QueueTriggeredFunction")]
public static void Run([ServiceBusTrigger("demofunctionqueue", Connection = "queueconstring")]string[] myQueueItems, ILogger log)
{
    log.LogInformation("Received messages {count}", myQueueItems.Length);
    foreach (var myQueueItem in myQueueItems)
    {
        log.LogInformation($"C# ServiceBus queue trigger function processed message: {myQueueItem}");
    }
}

host.json to set the maximum number of messages per batch

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "batchOptions": {
        "maxMessageCount": 200
      },
      "messageHandlerOptions": {
        "maxConcurrentCalls": 1
      }
    }
  }
}