Timelake

Service Bus

Do the Azure setup first.

Devices
send_messages
AMQP / HTTPS
Service Bus queue
batch
Azure Function
ingest_service_bus
commit
lake container
Iceberg tables

Devices send messages to a queue, each one reading or a list of readings. The function collects up to 1,000 messages or 30 seconds' worth per call and commits them at once. When a call fails, Service Bus hands the messages out again. A message that fails 5 times moves to the queue's dead-letter queue, where you can inspect it.

export SB_NAMESPACE=timelake-$RANDOM
az servicebus namespace create -g $RG -n $SB_NAMESPACE -l $LOCATION --sku Basic
az servicebus queue create -g $RG --namespace-name $SB_NAMESPACE -n readings \
  --max-delivery-count 5 --lock-duration PT5M
export SB_SCOPE=$(az servicebus namespace show -g $RG -n $SB_NAMESPACE --query id -o tsv)
az role assignment create --role "Azure Service Bus Data Receiver" --scope $SB_SCOPE \
  --assignee-object-id $PRINCIPAL --assignee-principal-type ServicePrincipal
az functionapp config appsettings set -g $RG -n $APP --settings \
  ServiceBusConnection__fullyQualifiedNamespace=$SB_NAMESPACE.servicebus.windows.net

The 5-minute lock keeps a batch's messages from being handed out again while the function holds them.

Try it:

az role assignment create --role "Azure Service Bus Data Sender" --scope $SB_SCOPE \
  --assignee $(az ad signed-in-user show --query id -o tsv)
pip install azure-servicebus azure-identity
import json
import os
 
from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient, ServiceBusMessage
 
readings = [
    {"time": "2026-01-05T10:10:00Z", "device_id": "pump-1", "temp": 63.1, "humidity": 41},
    {"time": "2026-01-05T10:10:00Z", "device_id": "pump-2", "temp": 60.2, "humidity": 44},
]
client = ServiceBusClient(f"{os.environ['SB_NAMESPACE']}.servicebus.windows.net", DefaultAzureCredential())
with client, client.get_queue_sender("readings") as sender:
    sender.send_messages(ServiceBusMessage(json.dumps(readings)))

Devices without the SDK can send over HTTPS with Service Bus's REST API and a shared access signature.