Document Batch Processing with Azure Document Intelligence

Introduction: synchronous, asynchronous, and batch processing

Document Intelligence can be called in three ways, and they suit different workloads:

• Synchronous — send one document, block until the result comes back. Simple, but the caller holds an open connection for the full analysis time (seconds to tens of seconds per document) and is limited to one document per call. Fine for interactive scenarios such as “scan this receipt now”.
• Asynchronous — send one document, receive an operation ID, poll for the result. The caller no longer blocks, but it is still one document per request: the caller has to enumerate files, submit each one, track every operation ID, retry failures, and collect results itself. For a folder of a thousand documents that is a thousand submissions and a thousand polling loops of application code.
• Batch — point the service at a storage container and receive one operation ID for the whole set. The service enumerates the documents, runs the analyses in parallel on its side, writes the results straight back to storage, and reports per-document outcomes when done. The caller’s job shrinks to “start it, check it, record the outcome”.

Press enter or click to view image in full size

Event-Driven Document Processing with Azure AI Document Intelligence

Document processing workloads come in two shapes. In one, documents arrive in drops — a nightly export, a scanned mailbag, a migration — and the question is “is the whole set done?”. That shape is served by Document Intelligence’s analyzeBatch operation and a batch-oriented design. In the other, documents arrive one at a time, continuously: an invoice is emailed, a form is uploaded from a portal, a scanner in a branch office pushes a file. Each document is its own unit of work, should be processed as soon as it appears, and should be individually traceable afterwards. That is the shape this solution targets.

The design is event-driven: the act of storing a file is the trigger. Nothing polls, nothing is scheduled, and no client has to call “start”. The result for each document is available within seconds, and a record of every document — what it was, when it arrived, how long it took, what it cost, whether it succeeded — is kept in a queryable table.

Her’s the architecture we will build for our event-driven approach

Press enter or click to view image in full size

Press enter or click to view image in full size

Why a queue between Event Grid and the function

Event Grid can invoke a function directly. The queue adds one hop (well under a second) and buys four things: concurrency is set by the consumer (batchSize × instances) rather than by how fast events arrive, so a burst of 500 uploads does not become 500 simultaneous calls into Document Intelligence's rate limit; messages that fail repeatedly move to documents-poison automatically, giving a durable list of problem files; reprocessing is "put a message back on the queue"; and the backlog is visible with a single az storage message peek. For a trickle of documents the queue is optional; for bursts it is the difference between smooth processing and a storm of 429 retries.

Idempotency and retries

Event Grid delivers at least once, so the same event can arrive twice. The processor treats the pair (upload date, blob name, blob ETag) as the identity of a piece of work: if a succeeded row exists with the same ETag, the second delivery returns without calling Document Intelligence. A re-uploaded file has a new ETag and is processed again. Transient errors re-raise so the queue redelivers with exponential back-off, up to maxDequeueCount (5); on the fifth failure the message lands in documents-poison and the row records the last error. Permanent errors — corrupt file, unsupported content — skip retries and go straight to failed/.

Table Design

Press enter or click to view image in full size

“What was processed today?” is a single-partition query. “What happened to invoice.pdf?” is a cross-partition filter on RowKey, which Table Storage handles comfortably at this scale; the API returns the latest attempt. If per-name lookups come to dominate, add a second table keyed by name or move to Cosmos DB.

All the terraform scripts needed to create the service can be found here. We’ll be creating the following services.

• Resource group, storage account, three containers, queue, table.
• EventGrid system topic with a system-assigned identity, role assignment Storage Queue Data Message Sender, and a subscription filtered to subjectBeginsWith /blobServices/default/containers/source/ with a queue destination. Because the destination is a queue rather than an HTTP endpoint there is no validation handshake and no two-phase apply.
• Document Intelligence (FormRecognizer, S0) with a pre-step that purges a soft-deleted account of the same name if one exists, and diagnostic settings to Log Analytics.
• Log Analytics workspace and workspace-based Application Insights.
• Linux consumption plan and Function App; app settings carry the Document Intelligence endpoint and key, model, and cost rate card. zip_deploy_file points at an archive of app/ whose filename embeds a hash of the source, so code changes trigger a redeploy and unchanged code does not.

Here’s how our infrastructure code and application code is structured for development.

With this structure in place, let’s move to terraform directory and create required infrastructure

Pre-requisite for this execution is terraform v16.x

terraform --version
Terraform v1.16.2
on linux_amd64
+ provider registry.terraform.io/hashicorp/archive v2.8.1
+ provider registry.terraform.io/hashicorp/azurerm v4.81.0
+ provider registry.terraform.io/hashicorp/random v3.9.1

This is what we have in our tfvars

base_name = "docasync"       # prefix for every resource name; lowercase letters/numbers only
location  = "eastus2"        # region with Document Intelligence availability

docintel_model = "prebuilt-layout"

# Rate card for estimated cost (USD per 1,000 pages) — verify against current pricing for your region
cost_per_1000_pages = {
  "prebuilt-read"   = 1.50
  "prebuilt-layout" = 10.00
}

log_retention_days = 365


tags = {
  project     = "docasync"
  environment = "dev"
  costcenter  = "1234"
  owner       = "your.name@example.com"
}

app_dir = "../app"

Let’s kick start execution and take a quick coffe break

terraform init
terraform plan
# check to see is all is well and we can proceed with apply step
terraform apply

Post execution of this step you should see these following services in portal

Let’s upload sample files into storage account.

All these sample files are available in repository. Note: These sample files are AI generated

krishnan@LinuxThinkpad:~/Projects/azure/DocIntelligence/AsyncProcessing$ az storage blob upload -c source -f ./sample-docs/invoice-INV-2026-1042.pdf -n invoice-INV-2026-1042.pdf --connection-string "$CONN" -o none
Finished[#############################################################]  100.0000%
krishnan@LinuxThinkpad:~/Projects/azure/DocIntelligence/AsyncProcessing$ az storage blob upload -c source -f ./sample-docs/letter-brightleaf-billing-correction.pdf -n letter-brightleaf-billing-correction.pdf --connection-string "$CONN" -o none
Finished[#############################################################]  100.0000%
krishnan@LinuxThinkpad:~/Projects/azure/DocIntelligence/AsyncProcessing$ az storage blob upload -c source -f ./sample-docs/memo-onsite-documentation.pdf -n memo-onsite-documentation.pdf --connection-string "$CONN" -o none
Finished[#############################################################]  100.0000%
krishnan@LinuxThinkpad:~/Projects/azure/DocIntelligence/AsyncProcessing$ az storage blob upload -c source -f ./sample-docs/minutes-cedar-point-kickoff.pdf -n minutes-cedar-point-kickoff.pdf --connection-string "$CONN" -o none
Finished[#############################################################]  100.0000%

Pushing this file to “source” container will automatically trigger an event over to queue which in turn will trigger azure function’s process_document method. Give it a couple of mins.

@app.queue_trigger(arg_name="msg", queue_name="documents", connection="AzureWebJobsStorage")
def process_document(msg: func.QueueMessage) -> None:
    event = msg.get_json()
    if event.get("eventType") != "Microsoft.Storage.BlobCreated":
        return
    url: str = event["data"]["url"]
    name = unquote(url.split("/source/", 1)[1])
    if os.path.splitext(name)[1].lower() not in SUPPORTED:
        log.info("Skipping unsupported file %s", name)
        return

    src = blobs.get_blob_client("source", name)
    try:
        props = src.get_blob_properties()
    except ResourceNotFoundError:
        log.info("Blob %s already gone (processed by an earlier delivery?)", name)
        return

    started = now()
    partition = started.strftime("%Y-%m-%d")
    etag = props.etag.strip('"')

    # Idempotency: Event Grid delivers at least once; don't pay for the same blob version twice.
    try:
        existing = table.get_entity(partition, row_key(name))
        if existing.get("status") == "succeeded" and existing.get("etag") == etag:
            log.info("Duplicate event for %s, already processed", name)
            return
    except ResourceNotFoundError:
        pass

    row = {
        "PartitionKey": partition,
        "RowKey": row_key(name),
        "name": name,
        "size_bytes": props.size,
        "content_type": props.content_settings.content_type,
        "etag": etag,
        "uploaded": props.last_modified,
        "model": MODEL,
        "status": "processing",
        "started": started,
        "attempt": msg.dequeue_count,
    }
    table.upsert_entity(row, mode=UpdateMode.REPLACE)

    try:
        data = src.download_blob().readall()
        result = docintel.begin_analyze_document(MODEL, AnalyzeDocumentRequest(bytes_source=data)).result()

        result_path = f"{name}.json"
        blobs.get_blob_client("processed", result_path).upload_blob(json.dumps(result.as_dict()), overwrite=True)
        blobs.get_blob_client("processed", name).upload_blob(data, overwrite=True)
        src.delete_blob()

        pages = len(result.pages or [])
        finished = now()
        row.update(
            status="succeeded",
            finished=finished,
            duration_ms=int((finished - started).total_seconds() * 1000),
            pages=pages,
            tables=len(result.tables or []),
            key_value_pairs=len(result.key_value_pairs or []),
            estimated_cost=round(pages * RATES.get(MODEL, 0) / 1000, 4),
            content_preview=(result.content or "")[:500],
            result_path=f"processed/{result_path}",
            original_path=f"processed/{name}",
        )
    except Exception as exc:  # noqa: BLE001
        if is_transient(exc) and msg.dequeue_count < 5:
            row.update(status="retrying", error_message=str(exc)[:500])
            table.upsert_entity(row, mode=UpdateMode.MERGE)
            raise  # queue redelivers with back-off; poison queue after maxDequeueCount
        log.exception("Failed %s", name)
        code = getattr(getattr(exc, "error", None), "code", None) or type(exc).__name__
        try:
            blobs.get_blob_client("failed", name).start_copy_from_url(src.url)
            src.delete_blob()
        except Exception:  # noqa: BLE001
            log.exception("Could not move %s to failed/", name)
        finished = now()
        row.update(
            status="failed",
            finished=finished,
            duration_ms=int((finished - started).total_seconds() * 1000),
            error_code=code,
            error_message=str(exc)[:1000],
            failed_path=f"failed/{name}",
        )

    table.upsert_entity(row, mode=UpdateMode.REPLACE)
    emit("DocumentProcessed", name=name, status=row["status"], model=MODEL, pages=row.get("pages"),
         estimated_cost=row.get("estimated_cost"), duration_ms=row.get("duration_ms"),
         error_code=row.get("error_code"), attempt=msg.dequeue_count, size_bytes=props.size)

The process_document function is an event-driven queue trigger that processes blobs uploaded to the /source container. Upon receiving a BlobCreated event, it validates the file type, checks Azure Table Storage to ensure the exact blob ETag hasn't already been successfully processed, and initializes a tracking record marked as processing. It then downloads the blob, sends it to Azure AI Document Intelligence for layout analysis, and saves both the raw document and the extracted JSON result to the /processed container before deleting the original from /source.

On success, it updates Table Storage with metrics like page count, duration, estimated cost, and content previews. If a transient network or service error occurs and retry attempts are under five, it updates the state to retrying and re-raises the exception so the queue can back off and redeliver; otherwise, for permanent failures or exhausted retries, it moves the document to the /failed container and records the error details in Table Storage, concluding by emitting a structured telemetry log.

We should now be able to invoke GET end point for extracted text.

@app.route(route="documents/{*name}", methods=["GET"])
def get_document(req: func.HttpRequest) -> func.HttpResponse:
    name = req.route_params["name"]
    want_result = name.endswith("/result")
    if want_result:
        name = name[: -len("/result")]

    rows = list(table.query_entities(f"RowKey eq '{row_key(name)}'"))
    if not rows:
        return json_response({"error": "not found", "name": name}, 404)
    row = max(rows, key=lambda r: r.get("started") or datetime.min.replace(tzinfo=timezone.utc))

    if not want_result:
        return json_response(to_json(row))
    if row.get("status") != "succeeded":
        return json_response({"error": f"no result: status is {row.get('status')}"}, 409)
    data = blobs.get_blob_client("processed", f"{name}.json").download_blob().readall()
    return func.HttpResponse(data, mimetype="application/json")

The get_document endpoint acts as a retrieval service for document processing history and extracted outputs. When queried via GET /documents/{name}, it looks up the file in Azure Table Storage using its normalized path as the row key, selects the most recent attempt if multiple exist, and returns a sanitized JSON payload containing processing metadata like status, page counts, duration, and estimated cost (returning a 404 if no record is found). If the client appends /result to the URL path, the endpoint checks that the document status is succeeded (returning a 409 Conflict if it failed or is still processing) and directly downloads and streams the full Document Intelligence JSON extraction from the /processed blob container.

# This is Azure function endpoint taken from portal/TF output
API=https://docasync-func.azurewebsites.net/api/documents

curl -s "$API/invoice-INV-2026-1042.pdf?code=$KEY" | jq .
# Execution results
{
  "attempt": 1,
  "content_preview": "INVOICE\nAcme Network Services, Inc. 1500 Polaris Pkwy, Columbus, OH 43240 billing@acmenet.example | (614) 555-0142\nBill to\nBrightleaf Dental Group\n220 Elm Street, Suite 300, Dublin, OH 43017\nInvoice # INV-2026-1042 Date 2026-08-12 Due 2026-09-12 PO PO-8822\nQty\nDescription\nUnit price\nAmount\n1\nServer rack, 42U\n$1,249.00\n$1,249.00\n4\nInstallation labor (hr)\n$125.00\n$500.00\n4\nCat6 patch cable, 3 ft (25 pk)\n$64.50\n$258.00\n1\nCable management kit\n$34.90\n$34.90\nSubtotal\n$2,041.90\nSales tax (7.25%)\n$148.0",
  "content_type": "application/pdf",
  "duration_ms": 6332,
  "estimated_cost": 0.01,
  "etag": "0x8DF11CD68B3ADC3",
  "finished": "2026-09-13T19:30:19.926288+00:00",
  "key_value_pairs": 0,
  "model": "prebuilt-layout",
  "name": "invoice-INV-2026-1042.pdf",
  "original_path": "processed/invoice-INV-2026-1042.pdf",
  "pages": 1,
  "result_path": "processed/invoice-INV-2026-1042.pdf.json",
  "size_bytes": 2649,
  "started": "2026-09-13T19:30:13.593471+00:00",
  "status": "succeeded",
  "tables": 2,
  "uploaded": "2026-09-13T19:30:03+00:00"
}

curl -s "$API/letter-brightleaf-billing-correction.pdf?code=$KEY" | jq .
# Execution results
{
  "attempt": 1,
  "content_preview": "Brightleaf Dental Group 220 Elm Street, Suite 300 Dublin, OH 43017\nSeptember 4, 2026\nMs. Dana Whitfield Accounts Receivable Acme Network Services, Inc. 1500 Polaris Pkwy Columbus, OH 43240\nRe: Invoice INV-2026-1042 - request for corrected billing\nDear Ms. Whitfield,\nThank you for the prompt completion of the network refresh at our Elm Street office. On reviewing invoice INV-2026-1042, we noticed that the installation labor line reflects 14 hours, while our site log and the technician's sign-off ",
  "content_type": "application/pdf",
  "duration_ms": 7274,
  "estimated_cost": 0.01,
  "etag": "0x8DF11CD84D7A787",
  "finished": "2026-09-13T19:31:09.793517+00:00",
  "key_value_pairs": 0,
  "model": "prebuilt-layout",
  "name": "letter-brightleaf-billing-correction.pdf",
  "original_path": "processed/letter-brightleaf-billing-correction.pdf",
  "pages": 1,
  "result_path": "processed/letter-brightleaf-billing-correction.pdf.json",
  "size_bytes": 2401,
  "started": "2026-09-13T19:31:02.518754+00:00",
  "status": "succeeded",
  "tables": 0,
  "uploaded": "2026-09-13T19:30:50+00:00"
}

curl -s "$API/memo-onsite-documentation.pdf?code=$KEY" | jq .
# Execution results
{
  "attempt": 1,
  "content_preview": "MEMORANDUM\nTO:\nAll field technicians\nFROM:\nOperations\nDATE:\nSeptember 8, 2026\nSUBJECT:\nUpdated on-site documentation procedure\nEffective September 15, every completed job must include a signed customer acknowledgement, a photo of the finished rack or closet, and hours logged in the mobile app before leaving the site. Paper timesheets will no longer be accepted after September 30.\nThis change follows several billing disputes in Q2 where logged hours could not be reconciled against invoices. Quest",
  "content_type": "application/pdf",
  "duration_ms": 5219,
  "estimated_cost": 0.01,
  "etag": "0x8DF11CD99860130",
  "finished": "2026-09-13T19:31:36.705087+00:00",
  "key_value_pairs": 0,
  "model": "prebuilt-layout",
  "name": "memo-onsite-documentation.pdf",
  "original_path": "processed/memo-onsite-documentation.pdf",
  "pages": 1,
  "result_path": "processed/memo-onsite-documentation.pdf.json",
  "size_bytes": 2137,
  "started": "2026-09-13T19:31:31.485453+00:00",
  "status": "succeeded",
  "tables": 1,
  "uploaded": "2026-09-13T19:31:25+00:00"
}

curl -s "$API/minutes-cedar-point-kickoff.pdf?code=$KEY" | jq .
# Execution results
{
  "attempt": 1,
  "content_preview": "Project Kickoff - Cedar Point Network Refresh\nDate: 2026-09-02 | Location: Toledo site, Conference Room 1 | Minutes by: J. Okafor\nAttendees\nR. Alvarez (Cedar Point, Plant Manager), T. Nguyen (Cedar Point, IT), D. Whitfield (Acme, PM), S. Patel (Acme, Lead Engineer)\nDecisions\n1. Cutover will occur over the weekend of October 4-5 to avoid production impact.\n2. Cedar Point will provide badge access for two Acme technicians beginning September 22.\n3. Existing fiber runs will be re-certified rather t",
  "content_type": "application/pdf",
  "duration_ms": 4235,
  "estimated_cost": 0.01,
  "etag": "0x8DF11CDC770E497",
  "finished": "2026-09-13T19:33:18.180460+00:00",
  "key_value_pairs": 0,
  "model": "prebuilt-layout",
  "name": "minutes-cedar-point-kickoff.pdf",
  "original_path": "processed/minutes-cedar-point-kickoff.pdf",
  "pages": 1,
  "result_path": "processed/minutes-cedar-point-kickoff.pdf.json",
  "size_bytes": 2589,
  "started": "2026-09-13T19:33:13.945360+00:00",
  "status": "succeeded",
  "tables": 1,
  "uploaded": "2026-09-13T19:32:42+00:00"
}

We should be able to see all of these records in Azure portal as well.

Press enter or click to view image in full size

Complete code for Azure function with all 3 end points can be found here.

App Traces

Open Log analytics and execute this query for information on time of execution number of documents, pages and model used.

This Kusto (KQL) query aggregates daily operational metrics and processing costs for document analysis workloads, grouped by the specific Document Intelligence model used. It filters the AppTraces table for log entries matching the "docasync.DocumentProcessed" prefix, extracts the JSON payload following the log header, and calculates daily summaries including total documents processed, total pages analyzed, total estimated cost (rounded to two decimal places), failed document count, and the 95th percentile processing duration in milliseconds.

AppTraces
| where Message startswith "docasync.DocumentProcessed"
| extend e = parse_json(substring(Message, indexof(Message, " ") + 1))
| summarize documents = count(), pages = sum(toint(e.pages)),
            estimated_cost = round(sum(todouble(e.estimated_cost)), 2),
            failed = countif(e.status == "failed"),
            p95_duration_ms = percentile(toint(e.duration_ms), 95)
  by bin(TimeGenerated, 1d), model = tostring(e.model)

Press enter or click to view image in full size

Limitations and next steps

• Long documents. One invocation is capped at 10 minutes on the consumption plan. For very large PDFs, store the operation ID on the row and complete it on a timer, or use a Premium plan.
• Name collisions. Two uploads with the same name on the same day overwrite one row. Prefix uploads with a timestamp or GUID if that can happen.
• Identity everywhere. Replace the connection string and Document Intelligence key with managed identity before production.
• Cold start. If enquiry-API latency matters, move to Flex Consumption or Premium.
• Reporting. Point Power BI at the documents table for browsing, and at Log Analytics for trends; add a Cost Management export to reconcile estimated cost against actual spend monthly.

In our follow up article, we will see how use cases that trip into current limitations can be handled through batch process. Until then stay safe and keep reading.