Polars and Delta from a Durable Function

The need did not call for a distributed engine: a few hundred thousand rows per run, a simple transformation, a write into a lake table. Standing up a cluster for that is an expensive reflex: it starts in three minutes, costs while it warms up, and is justified by a volume you do not have. A library working in memory in one process is enough, and the serverless function hosting it costs what it consumes.

A Durable Function is made of two natures that cannot be mixed. The orchestrator describes the sequence and is **replayed from the beginning** at every resumption: everything in it must be deterministic. No clock, no randomness, and above all no I/O. Activities do the work and run once per successful call. Writing to the lake from the orchestrator produces as many writes as there are replays, and replay is normal operation, not an exception.

import azure.durable_functions as df

def orchestrator(context: df.DurableOrchestrationContext):
    # Sequencing and nothing else: this block is replayed on every resumption.
    days = yield context.call_activity("list_days", context.get_input())
    # The activities fan out; the wait itself stays deterministic.
    tasks = [context.call_activity("write_day", d) for d in days]
    written = yield context.task_all(tasks)
    return {"partitions": len(written)}

main = df.Orchestrator.create(orchestrator)

The activity reads, transforms and writes. The write mode is not a configuration detail: it is where what happens when the same activity is called twice gets decided. In append mode, a replay doubles the rows. Replacing a partition rewrites them, and that is the only one of the two that survives being replayed.

import os
import polars as pl

STORAGE = {
    "account_name": os.environ["ADLS_ACCOUNT"],
    # Managed identity rather than a key: nothing to rotate, nothing to leak.
    "use_azure_cli": "false",
    "azure_storage_use_emulator": "false",
}

def write_day(day: str) -> str:
    frame = (
        pl.scan_parquet(f"abfss://raw@{os.environ['ADLS_ACCOUNT']}.dfs.core.windows.net/{day}/*.parquet")
        .filter(pl.col("amount") > 0)
        .group_by("store", "item")
        .agg(pl.col("amount").sum().alias("revenue"))
        .with_columns(pl.lit(day).alias("day"))
        .collect()
    )
    frame.write_delta(
        "abfss://refined@account.dfs.core.windows.net/sales",
        mode="overwrite",
        storage_options=STORAGE,
        # Replace only the day's partition, not the table: that is what makes
        # the activity replayable without doubling the rows.
        delta_write_options={"partition_by": ["day"], "predicate": f"day = '{day}'"},
    )
    return day

One question this arrangement raises and does not solve: two activities writing the same table at the same instant. The format guarantees only one of the two transactions lands, the other failing on a conflict, which is the right behaviour provided the caller retries. That is why the activities are cut by partition rather than by batch: two distinct partitions contend over nothing, and the conflict becomes the exception instead of the normal regime.

This approach has a hard limit, better named than discovered: everything must fit in one function’s memory. Past that, either cut more finely, which has an end, or return to a distributed engine, and the economics invert. Nor do I say anything about maintaining the table: compacting small files and cleaning old versions are indispensable and are not done from the function that writes. They need their own periodic job, which this arrangement does not provide.

• Polars · write_delta
• Azure · Durable Functions en Python