Batch Processing¶

Summary¶

This chapter covers the batch processing layer for multi-file flatten and build operations. You will learn about the BatchFlattener and BatchBuilder classes, multi-file processing, source column injection and extraction, vertical and horizontal concatenation with column alignment, dedup operations (apply dedup, cascade dedup, first-wins strategy), and partition keys for splitting batch output.

Concepts Covered¶

Multi File Processing
Column Alignment
BatchFlattener Class
BatchBuilder Class
Source Column Injection
Source Column Extraction
Vertical Concat
Horizontal Concat
Apply Dedup
Cascade Dedup
First Wins Strategy
Partition Key

Prerequisites¶

Chapter 1: Foundation Concepts (Relational Tables, Primary Keys, YAML Configuration)
Chapter 2: Spec Model (BatchingConfig, DedupConfig, DedupTableConfig, SourceColumn)
Chapter 3: Flatten Engine (TreeFlattener Class)
Chapter 6: Build Engine (TreeBuilder Class)

Learning Objectives¶

After completing this chapter, you will be able to:

Process multiple files using BatchFlattener and BatchBuilder
Inject and extract source columns to track record provenance
Concatenate results vertically and horizontally with column alignment
Apply dedup and cascade dedup with the first-wins strategy
Configure partition keys for batch output splitting

Introduction¶

Real-world data processing rarely involves a single file. Healthcare systems export patient records across thousands of files. Financial systems produce daily transaction exports. Government agencies distribute data in periodic batches. The batch processing layer sits above the single-file flatten and build engines, orchestrating multi-file workflows with provenance tracking, deduplication, and partitioned reconstruction.

The batch layer addresses four challenges that single-file processing cannot:

Multi-file orchestration -- Globbing source files, iterating through them, and combining the results
Provenance -- Tracking which output row came from which source file
Deduplication -- Removing overlapping records that appear across multiple files
Partitioned rebuild -- Splitting combined tables back into per-file tree reconstructions

This chapter covers the full batch pipeline from file discovery through to partitioned tree reconstruction.

Multi File Processing¶

Multi-file processing is the foundation of the batch layer. The BatchFlattener accepts a glob pattern (configured in BatchingConfig.source or overridden at call time) and processes all matching files through the same TreeSpec.

The processing loop for each file:

Parse the file into a typed tree (using the configured parser, defaulting to parse_xml)
Create a fresh TreeFlattener instance (ensuring clean state per file)
Flatten the tree into tables
Extract source column values from the file path and parsed tree
Inject source columns into the appropriate table rows
Append the file's tables to the combined result

file_paths = sorted(Path(p) for p in glob.glob(source_glob))
for file_path in file_paths:
    tree = self._parser(file_path, self._root_class)
    flattener = TreeFlattener(self._spec, hooks=self._hooks, class_registry=self._class_registry)
    file_tables = flattener.flatten(tree)
    # ... inject source columns and combine

File paths are sorted to ensure deterministic processing order. This matters for dedup with first_wins strategy, where the file processed first determines which duplicate survives. Sorted order means alphabetically earlier files (often older exports) take priority.

Source Column Extraction¶

Source column extraction is the process of deriving metadata values from each file that can be injected into the output rows. The extract_source_columns function supports five extraction methods:

`from_` value	Extracts	Example output
`filename_stem`	File name without extension	`"patient_export_2024_01"`
`filename`	Full file name with extension	`"patient_export_2024_01.xml"`
`path`	Complete file path as string	`"/data/exports/patient_export_2024_01.xml"`
`file_stat`	File system statistic	`1706745600.0` (mtime)
`attribute`	Dot-path attribute on the parsed tree	`"2024-01-15"` (from tree.meta.export_date)

The file_stat method supports three statistics: mtime (last modification time), size (file size in bytes), and ctime (creation time). The attribute method walks a dot-delimited path on the parsed tree root using getattr chains, enabling extraction of metadata that lives inside the document itself.

def extract_source_columns(
    file_path: Path,
    source_column_specs: dict[str, SourceColumn],
    tree_root: Any,
) -> dict[str, Any]:
    result = {}
    for col_name, spec in source_column_specs.items():
        if spec.from_ == "filename_stem":
            result[col_name] = file_path.stem
        elif spec.from_ == "attribute":
            result[col_name] = _resolve_attr_path(tree_root, spec.path)
        # ... other extraction methods
    return result

Source Column Injection¶

Source column injection adds the extracted metadata values to table rows after flattening. The injection behavior is controlled by BatchingConfig.inject_into, which determines which tables receive the source columns.

Three injection modes:

"all" (default) -- Every table's rows receive the source columns
"root_only" -- Only the root table's rows receive source columns
inject_into: [table_a, table_b] -- Only the listed tables receive source columns

The root_only mode is particularly useful when the source column is also the partition key. If source_file is injected into all tables, the BatchBuilder can partition child tables directly by that column. But if source_file should only exist on root rows (to keep child tables clean), using root_only with proper FK relationships still enables correct partitioning through FK lineage.

Diagram: Batch Processing Pipeline¶

Batch Processing Pipeline

Type: workflow sim-id: batch-processing-pipeline
Library: p5.js
Status: Specified

Purpose: End-to-end animation of the batch processing pipeline: file glob, per-file flatten, source column injection, concatenation, dedup, and final output.

Components:

File system panel showing glob pattern matching multiple files
Per-file processing animation (parse, flatten, inject)
Concatenation visualization (tables stacking vertically)
Dedup visualization (duplicate rows being removed)
Final output tables with source columns visible
Step controls with stage labels

Interactions: Click each stage to see detailed input/output. Adjust the number of source files (2-5) to see how the pipeline scales. Toggle dedup on/off to compare output. Hover over rows to see their source file.

Learning Objective: (Apply, Bloom Level 3) Students can trace a record from its source file through the batch pipeline to its final position in the output tables.

Column Alignment¶

Column alignment is a post-processing step that ensures all rows in each table have the same set of columns. Different source files may produce rows with different column sets -- one file's tree might have an optional field present while another file's tree has it absent. After concatenation, the same table can contain rows with different column keys.

The _align_columns function resolves this by union-filling missing columns with None:

def _align_columns(tables: dict[str, list[dict[str, Any]]]) -> None:
    for _table_name, rows in tables.items():
        if not rows:
            continue
        all_keys: set[str] = set()
        for row in rows:
            all_keys.update(row.keys())
        for row in rows:
            for key in all_keys:
                row.setdefault(key, None)

This function runs after concatenation and dedup, ensuring the final output is schema-consistent. Without alignment, downstream consumers (Polars, PyArrow) would fail when trying to create DataFrames from rows with inconsistent column sets.

The alignment is non-destructive: it only adds missing columns with None values, never removing or modifying existing values. The operation modifies rows in-place for efficiency.

Vertical Concat¶

Vertical concatenation (the default BatchingConfig.concat mode) appends rows from each file's tables to the combined result. This is the natural concatenation mode for batch processing: each file contributes rows to the same set of tables, and the combined output contains all rows from all files.

The implementation is straightforward list extension:

for table_name, rows in file_tables.items():
    if table_name not in combined:
        combined[table_name] = []
    combined[table_name].extend(rows)

Vertical concatenation assumes that all files produce the same set of tables with compatible schemas. When schemas differ (different optional fields present), column alignment resolves the inconsistency after concatenation.

After vertical concatenation, a table that had 100 rows per file across 50 files will contain 5,000 rows. This is where dedup becomes important -- if the same patient appears in multiple files, the combined table will have duplicate rows.

Horizontal Concat¶

Horizontal concatenation is an alternative mode where each file's output adds new columns rather than new rows. This is useful when different files represent different aspects of the same entities (e.g., one file has demographics, another has lab results, for the same set of patients).

In horizontal mode, tables are joined rather than stacked. The join key (typically the primary key) aligns rows across files so that each entity gets additional columns from each file. This requires that the same entities appear in every file, identified by matching key values.

Horizontal concatenation is less common than vertical in practice, but it supports use cases where data is partitioned by attribute rather than by entity.

BatchFlattener Class¶

The BatchFlattener class orchestrates the complete batch flatten pipeline. It combines file globbing, parsing, per-file flattening, source column injection, concatenation, and dedup into a single flatten_batch method call.

class BatchFlattener:
    def __init__(
        self,
        spec: TreeSpec,
        *,
        parser: Optional[Callable] = None,
        hooks: Optional[HookRegistry] = None,
        class_registry: Optional[dict[str, type]] = None,
        root_class: Optional[type] = None,
    ):

The constructor accepts:

spec -- The TreeSpec with batching configuration
parser -- A callable that takes (path, root_class) and returns a tree (defaults to parse_xml)
hooks -- Optional hook registry passed to each per-file TreeFlattener
class_registry -- Maps node names to classes for introspection
root_class -- The root element's Python class for the parser

The flatten_batch method returns the same dict[str, list[dict[str, Any]]] format as the single-file TreeFlattener.flatten, making it a drop-in replacement in pipelines that previously processed single files.

After concatenation, the method applies dedup if configured. The dedup behavior depends on the cascade flag:

Non-cascade dedup -- Dedup only the target table (root table by default)
Cascade dedup -- Dedup the root table, then cascade through children maintaining referential integrity

Apply Dedup¶

The apply_dedup function performs row deduplication on a single table based on key columns and a strategy. It is the atomic dedup operation that both standalone and cascade dedup invoke.

The function supports three strategies:

first_wins -- Keep the first occurrence of each key; discard subsequent duplicates
last_wins -- Keep the last occurrence of each key; overwrite previous duplicates
error -- Raise a ValueError if any duplicates are found

def apply_dedup(rows: list[dict[str, Any]], config: DedupConfig) -> list[dict[str, Any]]:
    if config.strategy == "first_wins":
        seen_keys: set[tuple] = set()
        result = []
        for row in rows:
            k = tuple(row.get(c) for c in config.key)
            if k not in seen_keys:
                seen_keys.add(k)
                result.append(row)
        return result

The key is constructed as a tuple of column values from the DedupConfig.key list. For a config with key: [patient_id], the key for each row is a single-element tuple like ("P001",). For composite keys like key: [patient_id, encounter_date], the tuple has multiple elements.

First Wins Strategy¶

The first-wins strategy is the most common dedup approach in mountainash-treespec. It preserves the chronologically earliest version of each record, which aligns with the "source of truth is the first export" assumption common in batch data processing.

Because files are processed in sorted order, "first" means the alphabetically earliest filename. In naming conventions like export_2024_01.xml, export_2024_02.xml, this means January's version wins over February's when both contain the same patient.

The implementation is a single-pass algorithm with \( O(n) \) time and \( O(k) \) space where \( k \) is the number of unique keys. It scans rows in order, maintaining a set of seen keys. The first row with a given key is kept; subsequent rows with the same key are discarded.

For the inverse "last_wins" strategy, the implementation uses a dictionary that overwrites on each occurrence, naturally keeping only the last. The final list is constructed from dictionary values.

Cascade Dedup¶

Cascade dedup extends single-table dedup to maintain referential integrity across the entire table hierarchy. When duplicate root rows are removed, their child rows (linked by FK) become orphaned. Cascade dedup removes these orphans and then recursively deduplicates and prunes at each child level.

The cascade algorithm:

Dedup the root table using the configured key and strategy
Compute the surviving key set (all key values that remain after dedup)
For each child table in the FK map: a. Remove rows whose FK value is not in the parent's surviving key set (orphan removal) b. Resolve the child's dedup key: per-table override, then primary_key, then all-columns fallback c. Dedup the child table d. Recurse into grandchildren using the child's dedup key as the next parent key

def apply_cascade_dedup(
    tables: dict[str, list[dict[str, Any]]],
    dedup_config: DedupConfig,
    fk_map: dict[str, str | list[str]],
    root_table: str,
    primary_keys: dict[str, list[str]] | None = None,
) -> dict[str, list[dict[str, Any]]]:

The function returns a new dictionary (it does not mutate the input), making it safe to use in pipelines where the original tables are needed for comparison or auditing.

Cascade and Error Strategy

The spec validator prevents combining cascade: true with strategy: "error". Cascade dedup removes rows as part of its orphan-pruning step, which contradicts the "error on any duplicate" semantics. Use first_wins or last_wins with cascade mode.

Diagram: Cascade Dedup Visualization¶

Cascade Dedup Visualization

Type: microsim sim-id: cascade-dedup-viz
Library: p5.js
Status: Specified

Purpose: Step-through animation showing cascade dedup operating across three table levels (patients, encounters, diagnoses), highlighting orphan removal and per-table dedup at each level.

Components:

Three table panels (patients, encounters, diagnoses) with FK links visualized
Duplicate rows highlighted in red before removal
Orphan rows highlighted in orange
Surviving rows in green
Step counter showing cascade depth
Animation: duplicates removed, orphans found, removed, child dedup applied, recurse

Interactions: Step through the cascade one level at a time. Click any row to see its FK relationships. Toggle between "before" and "after" views at each level. Adjust the number of duplicates to see different cascade patterns.

Learning Objective: (Apply, Bloom Level 3) Students can trace the cascade dedup algorithm through multiple table levels and predict which rows survive.

Partition Key¶

The partition key is the mechanism that enables the BatchBuilder to split combined tables back into per-file partitions for individual tree reconstruction. It is declared in BatchingConfig.partition_key and typically corresponds to a source column that identifies the originating file.

The BatchBuilder.build_batch method uses the partition key to:

Identify unique partition values from the root table
Filter root rows by partition value
For each child table, filter rows by FK lineage (not by partition key directly) to correctly handle inject_into="root_only" scenarios
Build one tree per partition using the standard TreeBuilder.build method

The partitioning logic is smarter than a simple column filter. It uses the spec's FK map to determine which child rows belong to each partition through FK relationships rather than requiring the partition key to exist on every table. This means a root row with source_file="export_01" correctly claims its encounters through the patient_id FK even when encounters do not have a source_file column.

for table_name, rows in tables.items():
    if table_name == root_table:
        continue
    join_key = fk_map.get(table_name)
    if join_key is not None:
        # Filter by FK lineage
        root_key_set = {tuple(row.get(c) for c in join_cols) for row in partition_root_rows}
        filtered = [row for row in rows if tuple(row.get(c) for c in join_cols) in root_key_set]
    else:
        # Fallback: filter by partition_key column if present
        filtered = [row for row in rows if row.get(partition_key) == pval]

BatchBuilder Class¶

The BatchBuilder class is the inverse of BatchFlattener. Where BatchFlattener takes multiple files and produces combined tables, BatchBuilder takes combined tables and produces multiple trees -- one per partition.

class BatchBuilder:
    def __init__(
        self,
        spec: TreeSpec,
        hooks: Optional[HookRegistry] = None,
        class_registry: Optional[dict[str, type]] = None,
    ):

The build_batch method accepts the combined tables and an optional workers parameter for parallel processing:

def build_batch(
    self,
    tables: dict[str, list[dict[str, Any]]],
    workers: int = 1,
) -> list[tuple[Any, Any]]:

The return type is a list of (partition_key_value, tree) tuples. Each tuple pairs the partition identifier (e.g., "export_01") with the reconstructed tree for that partition.

Before building, the method strips batch-only columns (the partition key and any source columns) from the rows. These columns are metadata for the batch layer and should not leak into the reconstructed dataclass instances. The stripping uses a simple dictionary comprehension that filters out the batch column names.

When workers > 1, the builder uses ProcessPoolExecutor for parallel tree construction. Each partition is built in a separate process, requiring that all hooks, specs, and class types are picklable. The HookRegistry.validate_parallel method checks hook picklability before submitting work to the pool.

Diagram: BatchBuilder Partition and Rebuild¶

BatchBuilder Partition and Rebuild

Type: microsim sim-id: batch-builder-partition
Library: vis-network
Status: Specified

Purpose: Visualize how BatchBuilder partitions combined tables by partition key, filters child tables by FK lineage, strips batch columns, and rebuilds individual trees per partition.

Components:

Input: combined tables with multiple partitions shown in different colors
Partition step: tables split into groups by partition key
FK lineage: arrows showing how child rows are assigned to partitions via FK
Build step: each partition becoming a tree
Output: list of (key, tree) pairs

Interactions: Click a partition to isolate its data and see the rebuild in detail. Toggle between "partition key filter" and "FK lineage filter" modes to understand the difference. Hover over a row to see which partition claims it and why.

Learning Objective: (Analyze, Bloom Level 4) Students can explain why FK-lineage-based partitioning is necessary when inject_into is not "all".

Complete Batch Workflow¶

The complete batch workflow connects all the pieces. Here is a typical production configuration:

meta:
  name: patient-batch
  root_element: Patient
  propagate_keys: [patient_id]

nodes:
  Patient:
    table: patients
    primary_key: [patient_id]
    children:
      encounters:
        node: Encounter
        table: encounters
        cardinality: many
        join_key: patient_id
  Encounter:
    table: encounters

batching:
  source: "./exports/*.xml"
  source_columns:
    source_file:
      from: filename_stem
  inject_into: root_only
  partition_key: source_file
  dedup:
    key: [patient_id]
    strategy: first_wins
    cascade: true

The workflow processes as: glob all XML files, flatten each file with the patient spec, inject the source filename into root rows only, concatenate all results vertically, deduplicate patients by ID (first file wins) with cascade cleanup of encounter orphans, align columns, and produce the final combined tables. The BatchBuilder can then split these tables by source_file to reconstruct individual per-file trees.

Key Takeaways¶

Multi-file processing globs source files, flattens each independently, and combines results with deterministic sort-order processing.
Source column extraction derives metadata from file paths, file stats, and tree attributes; source column injection adds this metadata to output rows based on inject_into settings.
Vertical concatenation stacks rows from all files into shared tables; horizontal concatenation joins across files by key.
Column alignment ensures schema consistency by union-filling missing columns with None after concatenation.
Apply dedup removes duplicates per table using first_wins, last_wins, or error strategies based on configurable key columns.
Cascade dedup maintains referential integrity by removing orphaned child rows and recursing through the FK hierarchy.
Partition key enables the BatchBuilder to split combined tables back into per-file partitions using FK-lineage-based filtering.
The BatchBuilder supports parallel tree reconstruction via ProcessPoolExecutor when hooks and classes are picklable.