[Python] How to consume million of messages from azure service bus topic trigger using durable...

I use vscode to consume millions of messages per day from Azure service bus topic and save it to ADLGen2 using Python. The function app is a premium app service plan.

Here is the project structure

/project-azure-function-app
│
├── /custom_package
│ └── __init__.py
│ └── functions.py # Contains my business logic (e.g., append_dataset, save_to_db, etc.)
│
├── function_app.py # Main entry point: Service Bus trigger
│
├── host.json # Function app configuration
├── requirements.txt # List of Python dependencies
└── local.settings.json # Local settings


In host.json, I use the default parameters. I am not sure whether there are some better alternatives in my case. I welcome your input and ideas

{
"version": "2.0",
"logging": {
"applicationInsights": {
"samplingSettings": {
"isEnabled": true,
"excludedTypes": "Request"
}
}
},
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
},
"extensions": {
"durableTask": {
"hubName": "orchestrator_function",
"maxConcurrentActivityFunctions": 10,
"maxConcurrentOrchestratorFunctions": 10
}
},
"functionTimeOut": "00:10:00"
}


function_app.py at the root directory contains the following code:

import azure.functions as func
import logging
from azure.storage.filedatalake import DataLakeServiceClient, FileSystemClient, DatalakeDirectoryClient, DataLakeClient
from custom_package.functions import save_to_db_activity, get_file_syst, arg_name, subscription_name, topic_name, connection, append_dataset, global_data

app = func.FunctionApp()


@app.service_bus_topic_trigger(arg_name, subscription_name, topic_name, connection)
@app.durable_client_input(client_name="client")
def servicebus_topic_trigger(azservicebus: func.ServiceBusMessage, client: df.DurableOrchestrationContext) -> None:
get_msg = azservicebus.get_body().decode('utf-8')
_ = await client client.start_new('orchestrator_function', None, get_msg)


async def orchestrator_function(context: df.DurableOrchestrationContext):
data = context.get_input()

# Call the activity function to save data to Data Lake.
await context.call_activity('save_to_db_activity', data)
return True


def save_to_db_activity(data: str) -> None:
append_dataset(data, False)
char_nb = 1000
if len(global_data) > char_nb:
// Data lake stuff to save the data into a file.
.....

// Reset global_data
append_dataset("", True)


/custom_package/functions.py

....

global_data = ""


def append_dataset(data: str, reset: bool)-> None:
global global_data
global_data += data + '\n'

if reset:
global_data = ""


I get the following error:

Exception: Exception: The function 'orchestrator_function' doesn't exist, is disabled, or is not an orchestrator function. Additional info: No orchestrator functions are currently registered!


I though I registered the function orchestrator_function in the host.json.

I am open to different viewpoints.

Continue reading...