Skip to content

Case: Batch Processing Pipeline

Problem

Design a batch ingestion pipeline: ingest 10M CSV rows/day from S3/Blob into a normalized DB. Each row goes through validate → enrich → transform → persist. Restartable; idempotent; scalable.

Walkthrough

Clarify

  • 10M rows/day = ~115 rows/sec average; bursts to 1000+.
  • Files arrive via Blob upload (uploaded once).
  • Ingestion takes ~1-2h per file batch.
  • Failure mid-pipeline must be recoverable.
  • Each row independent (no cross-row order).
  • DB is Postgres; transform-heavy.

Architecture

[Blob: file uploaded]
  │
  │ Event Grid → trigger
  ▼
[Orchestrator] (Worker / IHostedService)
  │
  ├─→ Stream rows → Channel<Row>
  │
  └─→ Workers (N) read from Channel:
        ├── Validate (drop bad → DLQ)
        ├── Enrich (3rd-party API; cached)
        ├── Transform
        └── Bulk-insert → DB

Checkpointing: every N rows, persist offset.
Restart: resume from offset.

Channel-based pipeline

public async Task ProcessAsync(string blobUri, CancellationToken ct)
{
    var channel = Channel.CreateBounded<Row>(new BoundedChannelOptions(1000)
    {
        FullMode = BoundedChannelFullMode.Wait,
        SingleReader = false,
        SingleWriter = true
    });

    // Producer
    var producer = Task.Run(async () =>
    {
        await foreach (var row in StreamBlobRows(blobUri, ct))
            await channel.Writer.WriteAsync(row, ct);
        channel.Writer.Complete();
    }, ct);

    // Consumers (parallel)
    var consumers = Enumerable.Range(0, 8).Select(_ => Task.Run(async () =>
    {
        var batch = new List<Row>(100);
        await foreach (var row in channel.Reader.ReadAllAsync(ct))
        {
            if (!Validate(row)) continue;
            row = await EnrichAsync(row, ct);
            row = Transform(row);
            batch.Add(row);
            if (batch.Count >= 100)
            {
                await BulkInsertAsync(batch, ct);
                await CheckpointAsync(row.Offset, ct);
                batch.Clear();
            }
        }
        if (batch.Count > 0) await BulkInsertAsync(batch, ct);
    }, ct)).ToArray();

    await Task.WhenAll(consumers.Append(producer));
}

Backpressure

BoundedChannelOptions with capacity 1000. Producer awaits when full → upstream throttled.

Idempotency

  • Each row has natural key (SourceFile + RowNumber).
  • DB insert: INSERT ... ON CONFLICT (source_file, row_number) DO NOTHING.
  • Restart from checkpoint: rows already inserted skipped.

Checkpointing

CREATE TABLE IngestProgress (
    JobId UUID,
    LastOffset BIGINT,
    UpdatedAt TIMESTAMP
);

Update every 100-1000 rows. Restart reads LastOffset.

Bulk insert

// EF Core 8+ ExecuteUpdateAsync / ExecuteInsertAsync, or use:
await using var conn = new NpgsqlConnection(...);
await conn.OpenAsync(ct);

using var writer = conn.BeginBinaryImport(@"COPY targets (a, b, c) FROM STDIN (FORMAT BINARY)");
foreach (var row in batch)
{
    await writer.StartRowAsync(ct);
    await writer.WriteAsync(row.A, ct);
    await writer.WriteAsync(row.B, ct);
    await writer.WriteAsync(row.C, ct);
}
await writer.CompleteAsync(ct);

COPY is 10-100x faster than per-row INSERT.

Dead letter

Bad rows → DLQ blob / table:

/dlq/{jobId}/{date}/bad-rows.json

Manual review; potential reprocess.

Parallelism

  • 8 consumer tasks default.
  • Tune based on DB write capacity (don't overwhelm DB).
  • CPU-bound transforms: scale by core count.
  • I/O-bound enrichment: scale higher.

Enrichment with caching

public async Task<Row> EnrichAsync(Row row, CancellationToken ct)
{
    var enriched = await _hybridCache.GetOrCreateAsync(
        $"enrich:{row.Sku}",
        (state, ct) => state.api.LookupAsync(state.sku, ct),
        (api: _api, sku: row.Sku),
        cancellationToken: ct);

    return row with { Description = enriched.Description };
}

3rd-party API rate-limited → Polly resilience handler.

Throttling DB writes

private readonly SemaphoreSlim _dbSlots = new(4);

public async Task BulkInsertAsync(IList<Row> batch, CancellationToken ct)
{
    await _dbSlots.WaitAsync(ct);
    try { /* insert */ }
    finally { _dbSlots.Release(); }
}

Max 4 concurrent inserts; protects DB.

Failure recovery

Worker crashes mid-batch:
  - Channel buffer drains; remaining rows wait.
  - Restart: reads LastOffset; resumes.
  - Already-inserted rows skipped via UPSERT.

Observability

private static readonly Meter _m = new("Ingest");
private static readonly Counter<long> _rows = _m.CreateCounter<long>("ingest.rows");
private static readonly Histogram<double> _batchLatency = _m.CreateHistogram<double>("ingest.batch.latency.ms");

Track: - Rows/sec. - Bad-row rate. - Enrichment latency. - DB insert latency. - Memory / GC.

Hosting

  • Azure Container Apps with KEDA scale on queue depth.
  • AKS for advanced workflow (Argo Workflows, Dagster).
  • Azure Functions Premium for event-trigger model.
  • Aspire AppHost for local dev orchestration.

For 10M rows/day: a single ACA instance per file type often enough.

Bigger scale options

  • Spark / Databricks: industry standard for huge data.
  • Azure Data Factory: orchestration; managed.
  • dbt: transforms in DB (post-load).

But for "10M rows/day" Channel-based .NET is fine.

Trade-offs

Choice Why Trade-off
Channel .NET-native; tight integration Less mature than Spark
Bulk COPY Fast Postgres-specific
Checkpoint Restartable Storage
Cached enrichment Fast Stale risk

What we'd skip

  • Spark for 10M/day — Channel is fine.
  • Microservices per stage — complexity not warranted.
  • Kafka unless multi-consumer.

Cross-references