Batch Processing for Azure Document Intelligence

Photo by Roger Starnes Sr on Unsplash

Batch processing with Azure AI Document Intelligence

This is a follow up from our previous article on asynchronous processing. The reason to treat batch as its own design rather than “async in a loop” is where the fan-out lives. With sync or async the application owns the loop, which brings back the classic problems the earlier iteration of this solution ran into: function execution time limits, concurrency throttling, and per-file bookkeeping. With batch the service owns the loop, so the application is small, stateless, and largely indifferent to batch size.

What Document Intelligence offers out of the box for batch

The analyzeBatch operation (API version 2024-11-30 and later, all prebuilt and custom models) provides:

• Container-based input — an azureBlobSource (container URL plus optional prefix) or an explicit file-list JSONL; up to 10,000 documents per request.
• Container-based output — a resultContainerUrl and resultPrefix; the service writes one <file>.ocr.json per input document, identical in shape to a single-document analysis result.
• One operation to track — a status of notStarted / running / succeeded / failed, a completion percentage, and, on completion, a details array with each document's status, its result URL, and any error, plus succeeded/failed/skipped counts.
• Idempotency control — an overwriteExisting flag so a re-run can either regenerate or skip documents that already have output.
• Identity-based storage access — the Document Intelligence resource’s managed identity can be granted rights on the storage account, removing the need to mint SAS tokens.
• SDK support — begin_analyze_batch_documents in the Azure SDKs, with a continuation token that lets a different process resume the operation later.

Two limits shape the design: per-document results are only exposed when the whole batch finishes, and the operation record is retained for 24 hours after completion, so the application must persist the outcome itself.

When should we consider batchAPI approach?

When to consider batch processing

• Work arrives as a set, not a stream. Nightly exports, month-end statements, a scanned archive, a migration, a backfill after a model change. The set is known before processing starts.
• The meaningful outcome is “the set is done.” Someone downstream waits for the whole batch — a reconciliation job, a report, a hand-off — rather than reacting to individual documents.
• Volume is large relative to your appetite for orchestration code. Hundreds to ten thousand documents per run. The service enumerates, parallelizes, retries internally, and writes results to storage; your code makes one call and later reads one status.
• Latency per document doesn’t matter. Minutes to complete a batch is fine; nobody is waiting on a specific file.
• You want to avoid function time limits and concurrency tuning. The long-running work happens inside Document Intelligence, so the caller can be a consumption-plan function, a Logic App, or a cron job.
• Operational simplicity and idempotent re-runs are valued. One operation ID per drop, overwriteExisting=false to skip already-done files, a prefix per run.

What we intend to build?

Azure services for batch processing

For our purpose, we will build following resources in Azure — Resource group, Application service plan, Azure function with 4 endpoints, document intelligence service, storage account, log analytics and application insights.

Azure function will process a batch of documents dropped into Azure Blob Storage through Azure AI Document Intelligence, with:

• an API call to start a batch or can be time-triggered,
• an API call to check its progress and final outcome,
• results delivered to a processed container and problem files to a failed container,
• a durable record of every batch that survives beyond the service’s own 24-hour retention.

The design favors minimal code (one Python file, ~150 lines) and minimal infrastructure (four Azure resources).

Execution flow

How it works?

Start a batch — POST /api/batches (optional body {"prefix": "2026-09-10/", "model": "prebuilt-read"})

Check a batch — GET /api/batches/{batch_id}

List batches — GET /api/batches returns id, status, timestamps and counters for every batch.

Sweep — a timer every 5 minutes runs the same refresh on any non-final batch, so results are captured even if nobody polls. Document Intelligence discards batch operation results 24 hours after completion; the sweep guarantees they are persisted well inside that window.

Here’s High level design for our code

The Python code implements an asynchronous batch document processing pipeline using Azure Functions, Azure Blob Storage, and Azure AI Document Intelligence.

Async Job Submission (POST /batches):

The start_batch HTTP POST endpoint initiates an asynchronous batch document processing operation by generating a timestamped, unique batch_id and extracting optional model and file prefix parameters from the incoming request body. It constructs an AnalyzeBatchDocumentsRequest pointing to input files in the /source blob container and specifying an isolated /processed/{batch_id}/ output destination. It then calls Document Intelligence's begin_analyze_batch_documents method, captures the operation poller's continuation_token for downstream tracking, and persists an initial tracking state JSON with a "Submitted" status into the /status container before immediately returning the full tracking object to the client with an HTTP 202 Accepted status.

Note: I made start_batch as a http triggered endpoint for ease of article. In real world this can be time triggered or event triggered.

@app.route(route="batches", methods=["POST"])
def start_batch(req: func.HttpRequest) -> func.HttpResponse:
    body = req.get_json() if req.get_body() else {}
    batch_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6]
    model = body.get("model", MODEL)

    request = AnalyzeBatchDocumentsRequest(
        azure_blob_source=AzureBlobContentSource(container_url=f"{ACCOUNT_URL}/source", prefix=body.get("prefix")),
        result_container_url=f"{ACCOUNT_URL}/processed",
        result_prefix=f"{batch_id}/",
        overwrite_existing=True,
    )
    poller = docintel.begin_analyze_batch_documents(model_id=model, body=request)

    batch = {
        "batch_id": batch_id,
        "status": "Submitted",
        "model": model,
        "request": request.as_dict(),
        "continuation_token": poller.continuation_token(),  # lets a later invocation resume this operation
        "created": now(),
        "last_checked": None,
        "finished": None,
        "total": None,
        "results_written": 0,
        "succeeded": 0,
        "failed": 0,
        "skipped": 0,
        "files": [],
    }
    save_status(batch)
    return json_response(batch, 202)

Status & Progress Tracking (GET /batches/{id} & GET /batches):

The get_batch HTTP GET endpoint retrieves and updates the real-time processing status of a specific batch operation identified by its URL path parameter. It loads the existing execution state JSON from the /status storage container (returning an HTTP 404 error if the record does not exist) and passes it through the internal refresh function. This function uses the batch's saved continuation token to poll Document Intelligence, calculates progress by counting output files in the /processed container, handles file post-processing upon job completion, saves the updated state back to blob storage, and finally returns the refreshed batch record to the client.

Get Krishnan Sriram’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The list_batches HTTP GET endpoint provides a high-level summary overview of all historical and active batch operations stored in the system. It enumerates all tracking files within the /status blob container, loads their corresponding JSON records, and projects a curated subset of key metrics—including batch_id, overall status, creation and completion timestamps, total written outputs, and file success/failure/skip counts. The endpoint then sorts this collection in descending chronological order based on the creation timestamp before returning the complete list to the caller as an HTTP 200 JSON response.

@app.route(route="batches/{batch_id}", methods=["GET"])
def get_batch(req: func.HttpRequest) -> func.HttpResponse:
    batch = load_status(req.route_params["batch_id"])
    if not batch:
        return json_response({"error": "not found"}, 404)
    return json_response(refresh(batch))


@app.route(route="batches", methods=["GET"])
def list_batches(req: func.HttpRequest) -> func.HttpResponse:
    keys = ("batch_id", "status", "created", "finished", "results_written", "succeeded", "failed", "skipped")
    items = [{k: b.get(k) for k in keys} for b in all_batches()]
    return json_response(sorted(items, key=lambda x: x["created"], reverse=True))

Automated Cleanup Sweep (sweep timer):

The sweep function serves as an automated background reconciliation job that prevents batch execution state from being lost due to service retention limits. Executing on an Azure Functions timer trigger every five minutes, it retrieves all persisted batch tracking records from the /status storage container and iterates through any job whose status has not reached a final terminal state (Completed, CompletedWithErrors, or Failed). For each active or unfinalized batch, it invokes the refresh function to poll Azure AI Document Intelligence using its saved continuation token, update file counts, perform post-processing housekeeping, and persist the updated status back to storage—ensuring batches are fully finalized even if external clients never query the GET /batches/{batch_id} endpoint before Document Intelligence purges its 24-hour temporary operation history.

@app.timer_trigger(schedule="0 */5 * * * *", arg_name="timer", run_on_startup=False)
def sweep(timer: func.TimerRequest) -> None:
    for b in all_batches():
        if b["status"] not in FINAL:
            refresh(b)

Post-Processing & Lifecycle (finalize_file):

The finalize_file function handles the post-processing housekeeping for individual documents within a completed batch based on their Document Intelligence processing outcome. Extracted from the batch operation details, it checks each file's status: if processing succeeded, it identifies the generated OCR output in the /processed container and deletes the original file from the /source container to clean up storage. If the document failed, it copies the original file to a structured /failed/{batch_id}/ location, uploads a sidecar .error.json file containing the specific failure details, and removes the source file; skipped files are left intact in /source to prevent data loss. Any exceptions encountered during file relocation or deletion are safely caught and recorded as a housekeeping_error to ensure storage cleanup failures never overwrite or obscure the primary result from Document Intelligence.

• Succeeded: Confirms output in /processed and deletes the source file from /source.
• Failed: Copies source files to /failed/{batch_id}/, creates a {filename}.error.json sidecar log, and removes the original from /source.
• Skipped: Retains the source file in /source to avoid data loss.

def finalize_file(batch: dict, d) -> dict:
    """d is an AnalyzeBatchResultDetail: source_url, result_url, status, error."""
    name = unquote(d.source_url.split("/source/", 1)[1])
    error = d.error.as_dict() if d.error else None
    f = {"name": name, "status": d.status}
    src = blobs.get_blob_client("source", name)
    try:
        if d.status == "succeeded":
            f["output"] = unquote(d.result_url.split("/processed/", 1)[1])
            src.delete_blob()
        elif d.status == "failed":
            f["error"] = error
            dest = f"{batch['batch_id']}/{name}"
            blobs.get_blob_client("failed", dest).start_copy_from_url(https://medium.com/src.url)
            blobs.get_blob_client("failed", dest + ".error.json").upload_blob(json.dumps(error, indent=2), overwrite=True)
            src.delete_blob()
        else:  # skipped: result already existed; leave the source file alone
            f["error"] = error
    except Exception as exc:  # noqa: BLE001 — never let housekeeping hide the DI outcome
        logging.exception("Post-processing failed for %s", name)
        f["housekeeping_error"] = str(exc)[:300]
    return f

You can find complete code for this execution here

Here’s how we can see Overall status for execution to understand the cost of operation for ADI.

$:> CONN=$(az storage account show-connection-string -n docbatchyhy8kz -g rg-docbatch -o tsv)
$:> BATCH=20260912-201210-5f911f
$:> mkdir -p /tmp/$BATCH
$:> az storage blob download-batch -s processed --pattern "$BATCH/*" -d /tmp --connection-string "$CONN" -o none
$:> for f in /tmp/$BATCH/*.ocr.json; do
  printf "%-50s %3d pages\n" "$(basename "$f" .ocr.json)" "$(jq '.analyzeResult.pages | length' "$f")"
done | tee /dev/stderr | awk '{s+=$2} END {print "total:", s, "pages"}'

# Execution Status
Finished[#############################################################]  100.0000%0.0000%000%
delivery-note-DN-77421.png                           1 pages
invoice-INV-2026-1041.pdf                            1 pages
invoice-INV-2026-1042.pdf                            1 pages
invoice-INV-2026-1043.pdf                            1 pages
letter-brightleaf-billing-correction.pdf             1 pages
memo-onsite-documentation.pdf                        1 pages
minutes-cedar-point-kickoff.pdf                      1 pages
purchase-order-PO-8832.pdf                           1 pages
receipt-maple-main-A1734.pdf                         1 pages
report-q2-2026-operations.pdf                        2 pages
service-request-SR-20260905-017.pdf                  1 pages
total: 12 pages

Although Log Analytics is connected to ADI, you may not see page level information in there

Log Analytics only has what something emits into it: the function’s request logs (no per-file data), Document Intelligence’s request logs (one entry per API call, so one per batch), and its metrics (page totals per time bucket for the whole resource, not per file). Per-document detail has to come from the function, since it’s the only component that knows both the batch and the file names.

Batch API — what enterprises and teams should be aware of

• Operation records vanish 24 hours after completion. Without a persistence step (our status/ blob and sweep), a weekend batch is unauditable by Monday.
• No per-document visibility until the whole batch completes. A 5,000-document batch that stalls gives you a percentage and nothing else; plan monitoring around result-file counts and time-based alerts.
• No cancellation of a running batch and no partial re-run of failed files inside it; failures are re-submitted as a new batch. Keep batches to a size you’re comfortable re-running.
• Snapshot semantics and overlapping prefixes are the most common source of double-processing. Enforce “one prefix per drop” as a team convention.
• The batch shares the resource’s throughput quota with any interactive traffic. Separate resources for batch and real-time workloads prevent a bulk job from degrading a customer-facing feature.
• Result files are large. Set lifecycle policies on the results container or storage grows silently.

Cost and governance

• Billing is per page at the resource meter; nothing ties spend to a batch, a team, or a business unit. Attribution has to be built (page counts per batch, tags per resource, one resource per cost center where needed).
• Re-runs with overwriteExisting=true re-bill every page. Make false the default and require an explicit override.
• Free tier limits and regional quotas can quietly cap throughput; teams onboarding new workloads should check quota before a large first run.

Model lifecycle

• One model per batch; mixed document types need routing first.
• Custom models are resource- and region-bound. Promoting a model between environments means using the copy operation, and batch code must reference the right model ID per environment.
• API version pinning matters: batch behavior has changed between preview and GA versions, and an SDK upgrade can move the default version underneath you.

HA and DR for this solution

HA & DR implementation

Document Intelligence is a regional service with no built-in geo-replication or automatic failover. Everything else in the design can be made resilient; this is the part that needs a decision.

Availability within a region

• Storage: use ZRS (zone-redundant) so source, results, status, and the queue survive a zone outage.
• Function App: consumption plans span zones by default; nothing to do.
• Document Intelligence: the SLA covers the service; a full regional outage is the DR case below.

Disaster recovery (regional outage)

• Data: make the storage account GRS or RA-GRS. Source documents, results, and status records then exist in the paired region. Note that the queue and table also replicate, but geo-failover of a storage account is a manual, account-wide operation with some data-loss window (RPO in minutes).
• Compute and service: keep a second Document Intelligence resource in the paired region. For prebuilt models that’s just a resource; for custom models, copy them to the secondary and keep the IDs in configuration. The Function App and Event Grid topics are recreated from Terraform in minutes — your RTO is essentially “time to terraform apply in region B plus storage failover", which is why infrastructure-as-code matters here more than anything else.
• In-flight batches: a batch running in the failed region is lost — operation records are regional. Because results were written to storage as they completed and the design uses overwriteExisting=false on re-runs, restarting the same batch in the secondary region processes only what didn't finish. That is the recovery procedure: fail over storage, re-point configuration, resubmit any batch not in a final status.
• In-flight per-document work (the async design): messages in the queue replicate with the storage account; after failover, the secondary function app drains them. The ETag-based idempotency check prevents double-processing of anything that completed just before the outage.

Active/active versus active/passive
Active/passive (secondary provisioned but idle, or provisioned on demand from Terraform) is the sensible default: it costs little, and document processing tolerates an RTO of an hour. Active/active — two regions each processing part of the load — mostly buys throughput, not resilience, and complicates the “which region processed this file” question in reporting.

We covered a lot in this combined section. Hope this helped your way to build better resilient systems. Stay safe and keep reading.