Definition

Batch processing is the execution of a collection of data operations as a cohesive unit, processed sequentially or in parallel without requiring interactive user intervention during execution. Unlike stream processing (which handles individual events in real-time), batch processing collects data into groups and processes them together, optimizing throughput at the cost of latency. Batch processing is the foundation of ETL (Extract, Transform, Load) pipelines, scheduled data synchronization, bulk data transformations, and intelligence collection workflows.

The concept originates from mainframe computing where "batches" of punched cards were submitted for overnight processing. In modern distributed systems, batch processing encompasses everything from MapReduce jobs processing petabytes of data to BEAM Task pools executing thousands of concurrent operations within a single Erlang VM. The critical distinction from stream processing is that batch operations have a defined beginning and end -- the input dataset is finite and known before processing starts.

In the Prismatic Platform, batch processing drives the DD pipeline's entity loading, OSINT tool bulk execution, scheduled data synchronization, quality gate analysis across the umbrella applications, and bulk intelligence report generation. The platform leverages Elixir's Task.async_stream/3, Stream.chunk_every/2, and supervised task pools to achieve fault-tolerant parallel batch execution with configurable concurrency limits and backpressure mechanisms.

Core Concepts

Batch Processing Taxonomy

Concept	Description	BEAM Primitive	Prismatic Implementation
Chunk	Fixed-size subset of the input dataset	`Stream.chunk_every/2`	DD entity groups (100-500 items)
Worker	Process executing chunk operations	`Task` / `GenServer`	Supervised task per chunk
Concurrency	Number of simultaneous workers	`max_concurrency` option	`System.schedulers_online() * 2`
Backpressure	Flow control preventing overwhelm	GenStage demand	DD pipeline rate limiting
Timeout	Maximum time per chunk/batch	`:timeout` option	5 minutes per chunk default
Progress	Tracking completion percentage	Telemetry events	PubSub "dd:pipeline" updates
Retry	Handling transient failures	Custom retry logic	Exponential backoff per chunk
Idempotency	Safe re-execution of operations	Content hashing	DD entity upsert by external_id

Batch vs. Stream vs. Micro-Batch Processing

Characteristic	Batch	Stream	Micro-Batch	Prismatic Usage
Latency	Minutes to hours	Milliseconds to seconds	Seconds to minutes	DD pipeline (batch), LiveView (stream)
Throughput	Very high	Moderate	High	Bulk entity loading, scheduled fetches
Complexity	Lower	Higher	Medium	Broadway for hybrid approaches
State	Stateless per batch	Stateful across events	Window-based state	GenServer-managed batch state
Error Handling	Retry entire batch or skip items	Per-event error handling	Per-window retry	Configurable per pipeline stage
Input Boundary	Finite, known dataset	Infinite, unbounded	Finite windows of stream	DD: finite groups; OSINT: scheduled windows
Ordering	Not guaranteed (parallel)	Preserved (sequential)	Window-ordered	Chunk ordering preserved within batch
Memory	Proportional to chunk size	Constant per event	Proportional to window	Controlled via chunk_size parameter

Batch Execution Patterns

Pattern	Description	OTP Implementation	When to Use
Sequential	Process items one at a time	`Enum.each/2`	Order-dependent operations, low volume
Parallel	Process items concurrently	`Task.async_stream/3`	Independent items, CPU-bound work
Chunked	Split into fixed-size chunks	`Stream.chunk_every/2`	Memory-constrained environments
Supervised	Fault-tolerant parallel	`Task.Supervisor.async_stream_nolink/3`	Production workloads, crash isolation
Demand-driven	Consumer pulls from producer	`GenStage` producer-consumer	Backpressure-sensitive pipelines
Broadway	Multi-stage with acknowledger	`Broadway` pipeline	Message queue consumption
Flow	Parallel data processing	`Flow` partitioned stages	Large dataset transformations

Backpressure Mechanisms

Mechanism	Description	Implementation	Trade-off
max_concurrency	Limit concurrent tasks	`Task.async_stream(enum, fn, max_concurrency: n)`	Simple but coarse-grained
GenStage demand	Consumer requests from producer	`handle_demand/2` callback	Fine-grained but complex
Token bucket	Rate-limited execution	Custom GenServer with token counter	Predictable rate, variable latency
Queue depth	Monitor queue size, pause producers	ETS counter + threshold check	Adaptive but requires tuning
Circuit breaker	Stop processing on repeated failures	State machine in GenServer	Prevents cascade failures

Technical Deep Dive

BEAM Process Model and Batch Processing

The BEAM VM provides unique advantages for batch processing that distinguish it from thread-based runtimes. Each Elixir process is lightweight (approximately 2KB initial memory), preemptively scheduled, and isolated -- a crash in one process cannot corrupt another. This makes the BEAM ideal for batch processing where thousands of concurrent operations must execute reliably.

The scheduler architecture matters directly for batch throughput. The BEAM runs one scheduler per CPU core by default, and each scheduler manages a run queue of processes. When Task.async_stream/3 spawns concurrent tasks, the BEAM distributes them across schedulers automatically. The max_concurrency parameter controls how many tasks are active simultaneously, preventing scheduler overload.

Reference-counted binaries (refc binaries) are critical for batch processing of large payloads. When a binary larger than 64 bytes is shared between processes via message passing, only a reference is copied -- the binary data itself lives in a shared heap. This means distributing large HTTP response bodies or CSV file chunks across worker tasks is nearly zero-cost in terms of memory copying.

Task.async_stream Internals

Task.async_stream/3 is the workhorse of batch processing in Elixir. It creates a stream that spawns tasks lazily as the stream is consumed, maintaining at most max_concurrency tasks running simultaneously. Internally, it uses a linked task that monitors completion and enforces the timeout.

The :ordered option (default true) controls whether results are emitted in input order or as they complete. For batch processing where order does not matter, setting ordered: false can improve throughput by avoiding head-of-line blocking where a slow task delays emission of completed results.

The :on_timeout option determines behavior when a task exceeds its timeout: :exit raises an error (default), while :kill_task kills the task and returns {:exit, :timeout}. For batch processing, :kill_task is preferred because it allows the batch to continue processing remaining chunks.

Chunking Strategies

Chunk size selection is a critical performance parameter with competing concerns:

Too small (1-10 items): Per-chunk overhead dominates. Process spawning, message passing, and result aggregation costs exceed processing costs.
Too large (10,000+ items): Memory pressure increases, timeout risk grows, and a single chunk failure wastes more work.
Optimal (100-1,000 items): Balances overhead against memory usage and failure blast radius. The exact optimum depends on per-item processing cost and available memory.

For database operations, chunk size should also consider transaction size limits and connection pool availability. A chunk of 500 inserts in a single transaction is typically faster than 500 individual inserts but slower than a bulk insert using Repo.insert_all/3.

Telemetry Integration

Every batch processing operation in the Prismatic Platform emits telemetry events at three granularities:

Batch level: [:prismatic, :batch, :started] and [:prismatic, :batch, :completed] with total counts and duration.
Chunk level: [:prismatic, :batch, :chunk, :completed] with per-chunk metrics for progress tracking.
Item level (optional): [:prismatic, :batch, :item, :failed] for individual failure tracking.

These events power real-time progress bars in the DD pipeline LiveView dashboard and feed into the platform's observability infrastructure for historical analysis.

Usage in Prismatic Platform

DD Pipeline: PrismaticDd.Client.fetch_group/1 and PrismaticDd.Loader.load_group/1 process entities in configurable batches with PubSub progress updates to the pipeline dashboard
OSINT Bulk Execution: Running multiple OSINT tools across a target list in batch mode with per-tool timeout enforcement and result aggregation
Quality Gates: mix quality.gates analyzes all umbrella apps in batch with parallel compilation checks
Promo Site Build: Zola processes content pages in batch during zola build for static site generation
Database Migrations: Bulk data migrations use batch processing with chunked inserts to avoid lock contention
DD Scheduler: Periodic batch execution of fetch+load per source group with cron-like scheduling
Intelligence Reports: Bulk generation of investigation reports across multiple entities with parallel OSINT enrichment
Entity Deduplication: Content-hash-based deduplication runs in batch across the full entity corpus
Telemetry Aggregation: Periodic batch aggregation of raw telemetry events into summary metrics

Code Examples

Supervised Batch Processor with Telemetry

defmodule PrismaticDd.BatchProcessor do
  @moduledoc """
  Supervised batch processor for the DD pipeline.

  Processes entity records in configurable chunk sizes with
  fault-tolerant parallel execution, progress tracking, and
  comprehensive telemetry instrumentation. Uses
  `Task.Supervisor.async_stream_nolink/4` to isolate chunk
  failures from the calling process.

  ## Architecture

  The processor splits input records into chunks, spawns supervised
  tasks for each chunk, aggregates results, and emits telemetry
  events at batch and chunk granularity. Failed chunks are tracked
  but do not halt the overall batch.

  ## Examples

      iex> records = [%{name: "Entity A"}, %{name: "Entity B"}]
      iex> {:ok, result} = PrismaticDd.BatchProcessor.process_batch(
      ...>   records,
      ...>   fn record -> {:ok, record} end,
      ...>   chunk_size: 1
      ...> )
      iex> result.processed
      2
  """

  require Logger

  @default_chunk_size 100
  @max_concurrency System.schedulers_online() * 2
  @default_timeout :timer.minutes(5)

  @typedoc "Result of a batch processing operation"
  @type batch_result :: %{
          processed: non_neg_integer(),
          failed: non_neg_integer(),
          skipped: non_neg_integer(),
          duration_ms: non_neg_integer(),
          errors: [map()],
          chunks_completed: non_neg_integer(),
          chunks_failed: non_neg_integer()
        }

  @typedoc "Options for batch processing"
  @type batch_opts :: [
          chunk_size: pos_integer(),
          max_concurrency: pos_integer(),
          timeout: pos_integer(),
          on_progress: (map() -> :ok) | nil,
          ordered: boolean()
        ]

  @doc """
  Process a batch of records with configurable parallelism.

  Splits `records` into chunks of `chunk_size`, processes each chunk
  in parallel using supervised tasks, and aggregates results.

  ## Options

    * `:chunk_size` - Number of items per chunk (default: #{@default_chunk_size})
    * `:max_concurrency` - Maximum parallel chunks (default: schedulers * 2)
    * `:timeout` - Per-chunk timeout in ms (default: 5 minutes)
    * `:on_progress` - Optional callback invoked after each chunk
    * `:ordered` - Whether to preserve input order (default: false)

  ## Examples

      iex> processor = fn record ->
      ...>   case Map.get(record, :valid, true) do
      ...>     true -> {:ok, record}
      ...>     false -> {:error, :invalid}
      ...>   end
      ...> end
      iex> {:ok, result} = PrismaticDd.BatchProcessor.process_batch(
      ...>   [%{name: "A", valid: true}, %{name: "B", valid: false}],
      ...>   processor
      ...> )
      iex> result.processed
      1
      iex> result.failed
      1
  """
  @spec process_batch(
          [map()],
          (map() -> {:ok, term()} | {:error, term()} | :skip),
          batch_opts()
        ) :: {:ok, batch_result()}
  def process_batch(records, processor_fn, opts \\ []) when is_list(records) do
    chunk_size = Keyword.get(opts, :chunk_size, @default_chunk_size)
    concurrency = Keyword.get(opts, :max_concurrency, @max_concurrency)
    timeout = Keyword.get(opts, :timeout, @default_timeout)
    on_progress = Keyword.get(opts, :on_progress)
    ordered = Keyword.get(opts, :ordered, false)

    total_count = length(records)
    start_time = System.monotonic_time(:millisecond)

    :telemetry.execute(
      [:prismatic, :batch, :started],
      %{total_count: total_count, chunk_size: chunk_size, concurrency: concurrency},
      %{source: :batch_processor}
    )

    chunks = Stream.chunk_every(records, chunk_size)

    initial_acc = %{
      processed: 0,
      failed: 0,
      skipped: 0,
      errors: [],
      chunks_completed: 0,
      chunks_failed: 0
    }

    results =
      chunks
      |> Task.Supervisor.async_stream_nolink(
        PrismaticDd.TaskSupervisor,
        fn chunk -> process_chunk(chunk, processor_fn) end,
        max_concurrency: concurrency,
        timeout: timeout,
        on_timeout: :kill_task,
        ordered: ordered
      )
      |> Enum.reduce(initial_acc, fn
        {:ok, chunk_result}, acc ->
          updated = %{
            acc
            | processed: acc.processed + chunk_result.processed,
              failed: acc.failed + chunk_result.failed,
              skipped: acc.skipped + chunk_result.skipped,
              errors: acc.errors ++ chunk_result.errors,
              chunks_completed: acc.chunks_completed + 1
          }

          if on_progress, do: on_progress.(updated)
          updated

        {:exit, reason}, acc ->
          Logger.warning("Batch chunk failed: #{inspect(reason)}",
            module: __MODULE__,
            chunk_size: chunk_size
          )

          %{
            acc
            | failed: acc.failed + chunk_size,
              chunks_failed: acc.chunks_failed + 1,
              errors: [%{type: :chunk_timeout, reason: reason} | acc.errors]
          }
      end)

    duration = System.monotonic_time(:millisecond) - start_time

    :telemetry.execute(
      [:prismatic, :batch, :completed],
      %{
        processed: results.processed,
        failed: results.failed,
        skipped: results.skipped,
        duration_ms: duration,
        chunks_completed: results.chunks_completed,
        chunks_failed: results.chunks_failed
      },
      %{source: :batch_processor}
    )

    {:ok, Map.put(results, :duration_ms, duration)}
  end

  @doc """
  Process a single chunk of records.

  Returns aggregated counts of processed, failed, and skipped items
  within the chunk.
  """
  @spec process_chunk([map()], (map() -> {:ok, term()} | {:error, term()} | :skip)) :: map()
  def process_chunk(chunk, processor_fn) when is_list(chunk) do
    Enum.reduce(chunk, %{processed: 0, failed: 0, skipped: 0, errors: []}, fn record, acc ->
      case processor_fn.(record) do
        {:ok, _} ->
          %{acc | processed: acc.processed + 1}

        {:error, reason} ->
          %{acc | failed: acc.failed + 1, errors: [{record, reason} | acc.errors]}

        :skip ->
          %{acc | skipped: acc.skipped + 1}
      end
    end)
  end
end

Demand-Driven Batch Producer with GenStage

defmodule PrismaticDd.BatchProducer do
  @moduledoc """
  GenStage producer that emits entity batches on demand.

  Implements backpressure by only fetching new batches when
  downstream consumers signal demand. This prevents memory
  exhaustion when the consumer (e.g., database writer) is slower
  than the producer (e.g., HTTP fetcher).

  ## Examples

      iex> {:ok, pid} = PrismaticDd.BatchProducer.start_link(
      ...>   source: :czech_registry,
      ...>   batch_size: 200
      ...> )
      iex> is_pid(pid)
      true
  """

  use GenStage

  require Logger

  @type state :: %{
          source: atom(),
          batch_size: pos_integer(),
          offset: non_neg_integer(),
          exhausted: boolean()
        }

  @doc "Start the batch producer linked to the calling process."
  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts) do
    GenStage.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl GenStage
  @spec init(keyword()) :: {:producer, state()}
  def init(opts) do
    state = %{
      source: Keyword.fetch!(opts, :source),
      batch_size: Keyword.get(opts, :batch_size, 100),
      offset: 0,
      exhausted: false
    }

    {:producer, state}
  end

  @impl GenStage
  def handle_demand(demand, %{exhausted: true} = state) do
    {:noreply, [], state}
  end

  def handle_demand(demand, state) when demand > 0 do
    count = demand * state.batch_size

    case fetch_entities(state.source, state.offset, count) do
      {:ok, entities} when entities == [] ->
        Logger.info("Batch producer exhausted source #{state.source}")
        {:noreply, [], %{state | exhausted: true}}

      {:ok, entities} ->
        {:noreply, entities, %{state | offset: state.offset + length(entities)}}

      {:error, reason} ->
        Logger.error("Batch fetch failed: #{inspect(reason)}")
        {:noreply, [], state}
    end
  end

  @spec fetch_entities(atom(), non_neg_integer(), pos_integer()) ::
          {:ok, [map()]} | {:error, term()}
  defp fetch_entities(source, offset, limit) do
    PrismaticDd.Client.fetch_group(source, offset: offset, limit: limit)
  end
end

Bulk Database Insert with Chunking

defmodule PrismaticDd.BulkInserter do
  @moduledoc """
  Efficient bulk database insertion with chunk-based transactions.

  Uses `Repo.insert_all/3` within transactions to batch database
  writes, avoiding the overhead of individual inserts while keeping
  transaction sizes bounded.

  ## Examples

      iex> entities = [%{name: "A", type: "company"}, %{name: "B", type: "person"}]
      iex> {:ok, result} = PrismaticDd.BulkInserter.bulk_insert(entities, chunk_size: 500)
      iex> result.inserted >= 0
      true
  """

  alias PrismaticDd.Repo
  alias PrismaticDd.Schemas.EntityRecord

  require Logger

  @default_chunk_size 500

  @doc """
  Insert entities in bulk using chunked transactions.

  Returns the total number of inserted and failed records.
  Each chunk is wrapped in its own transaction for isolation.
  """
  @spec bulk_insert([map()], keyword()) :: {:ok, map()} | {:error, term()}
  def bulk_insert(entities, opts \\ []) do
    chunk_size = Keyword.get(opts, :chunk_size, @default_chunk_size)
    now = DateTime.utc_now() |> DateTime.truncate(:second)

    results =
      entities
      |> Stream.map(&Map.merge(&1, %{inserted_at: now, updated_at: now}))
      |> Stream.chunk_every(chunk_size)
      |> Enum.reduce(%{inserted: 0, failed: 0}, fn chunk, acc ->
        case Repo.transaction(fn ->
               Repo.insert_all(EntityRecord, chunk,
                 on_conflict: :replace_all,
                 conflict_target: [:external_id, :source_slug]
               )
             end) do
          {:ok, {count, _}} ->
            %{acc | inserted: acc.inserted + count}

          {:error, reason} ->
            Logger.error("Bulk insert chunk failed: #{inspect(reason)}")
            %{acc | failed: acc.failed + length(chunk)}
        end
      end)

    {:ok, results}
  end
end

Common Pitfalls

Pitfall	Symptom	Root Cause	Solution
Unbounded concurrency	System overload, OOM	No `max_concurrency` limit	Always set `max_concurrency` to `schedulers * 2` or lower
Missing timeouts	Stuck batches, resource leaks	Infinite default timeout	Set `:timeout` and `:on_timeout` on every `async_stream`
Chunk size too large	Memory spikes, long GC pauses	Entire chunk loaded into process heap	Profile memory and reduce chunk size until GC pauses are acceptable
No progress tracking	Users think system is frozen	Silent long-running batches	Emit telemetry events at chunk boundaries, update LiveView via PubSub
Non-idempotent operations	Duplicate data on retry	Retrying inserts without upsert	Use `on_conflict` in `Repo.insert_all/3` or content hashing
Ignoring partial failures	Silent data loss	Discarding error results	Track failed items, log errors, return structured result with error details
Large binary retention	Memory bloat in long batches	Sub-binary references keeping parent alive	Use `:binary.copy/1` when extracting small fields from large HTTP responses
Blocking the caller	UI freeze during batch	Synchronous batch execution	Run batches in background Task, notify via PubSub on completion
Connection pool exhaustion	Timeouts on unrelated queries	Too many concurrent DB chunks	Set `max_concurrency` <= connection pool size
Missing supervision	Orphaned tasks after crash	Using `Task.async_stream` without supervisor	Use `Task.Supervisor.async_stream_nolink/4` in production

Best Practices

Size chunks appropriately: Start with 100-500 items per chunk. Profile memory usage and throughput to find the optimum for your workload. Database operations favor larger chunks (500-1000) while CPU-bound work favors smaller chunks (50-100).
Use Task.Supervisor.async_stream_nolink for production workloads: Unlike Task.async_stream, the supervised variant isolates task crashes from the calling process. A single chunk failure will not bring down the entire batch.
Implement progress tracking with telemetry: Emit [:prismatic, :batch, :chunk, :completed] events after each chunk. Wire these to PubSub for real-time LiveView dashboard updates. Users need visibility into long-running operations.
Handle partial failures explicitly: Decide upfront whether to retry failed items, skip them, or fail the entire batch. Return structured results with error details. Never silently discard failures.
Set timeouts per chunk, not per batch: A 5-minute timeout on a chunk is reasonable. A 5-minute timeout on a 10,000-item batch is too aggressive. Use on_timeout: :kill_task to continue processing remaining chunks.
Make operations idempotent: Use content hashing, ON CONFLICT clauses, or external_id-based upserts so that retrying a failed batch does not create duplicates.
Respect connection pool limits: Set max_concurrency to at most the database connection pool size. If your pool has 10 connections and you spawn 20 concurrent database-writing chunks, half will timeout waiting for connections.
Use Stream for lazy chunk generation: Stream.chunk_every/2 generates chunks lazily, avoiding loading the entire dataset into memory before processing starts. This is critical for datasets larger than available memory.
Implement backpressure for producer-consumer patterns: When the data source is faster than the sink, use GenStage or manual demand signaling to prevent unbounded queue growth. Monitor queue depth and pause producers when thresholds are exceeded.
Profile and benchmark with realistic data: Use Benchee to compare chunk sizes, concurrency levels, and processing strategies. What works for 1,000 items may not scale to 1,000,000. Test with production-representative volumes.

Aggregation -- combining batch-processed results into summary statistics
Cache -- caching batch results for subsequent access, ETS-backed
Pipeline -- multi-stage processing pipelines built from batch stages
CSV -- common batch data input format parsed from binary streams
GenServer -- process managing batch state and coordination
Telemetry -- instrumentation for batch progress and performance
ETS -- in-memory storage for batch intermediate results
Stream -- lazy enumeration underlying chunk generation
Connection Pool -- database connection management during batch writes
Configuration -- batch parameter configuration (chunk size, concurrency)
Data Pipeline -- end-to-end data flow incorporating batch stages
ETL -- extract-transform-load pattern built on batch operations

Connect & Contribute

Created by Tomas Korcak (korczis) | Open Source under GHL

GitHub | GitLab | LinkedIn | Contact
Developer Portal | Architecture | Meet the Creator

Batch Processing