Building High Performance ETL Pipelines for Multi-Gigabyte XML Files

Exporting Apple Health data yields a single zip file. Unzip it, and you find export.xml. If you have tracked data for a few years, this file easily exceeds 3 GB. Try opening it with a standard XML parser, and your script will crash. The operating system kills the process because it runs out of memory.

This happens because standard parsers build a Document Object Model (DOM) tree in memory. A 3 GB XML file can require 30 GB of RAM once parsed into memory objects.

Processing files of this size requires a streaming approach. You must parse, transform, and write data to your database in a single pass with a flat memory footprint.

The Memory Trap of DOM Parsing

Most high-level XML libraries default to tree parsing. They read the entire file, parse the tags, build parent-child relationships, and present a tree structure.

# Do not do this with multi-gigabyte files
import xml.etree.ElementTree as ET
 
tree = ET.parse('export.xml') # Crashes with large files
root = tree.getroot()

When Python parses an XML element into a tree node, it wraps the raw text in objects. A simple tag like <Record type="HKQuantityTypeIdentifierStepCount" value="120" /> takes up a few dozen bytes of text. In memory, the parser represents this with dictionary lookups, attribute string objects, and parent-child pointers. The memory overhead increases by a factor of ten.

To process a 3 GB export on a standard machine or a small cloud instance, you must keep memory usage constant, regardless of the file size.

Streaming XML with Pull Parsing

Streaming parsers do not load the whole document. They read the file byte by byte and emit events when they encounter start tags, attributes, text, or end tags.

There are two main streaming models:

1. Push Parsing (SAX): The parser controls the execution loop. You register callbacks for events.
2. Pull Parsing (StAX / iterparse): Your code controls the loop. You ask the parser for the next event.

Pull parsing is generally easier to write and maintain. Python's lxml library provides a tool for this called iterparse.

Here is the basic pattern for pull parsing:

from lxml import etree
 
def parse_records(file_path):
    # Only trigger events on the end tag of a 'Record' element
    context = etree.iterparse(file_path, events=('end',), tag='Record')
    
    for event, elem in context:
        # Process the element
        yield elem.attrib

This code looks like it streams, but it still leaks memory. iterparse keeps building the DOM tree in the background so you can access parent elements. If you run this on a 3 GB file, your memory usage will grow linearly until the process crashes.

You must explicitly clear the elements from memory after processing them.

def parse_records_safe(file_path):
    context = etree.iterparse(file_path, events=('end',), tag='Record')
    
    for event, elem in context:
        yield elem.attrib
        
        # Clear the element's content
        elem.clear()
        
        # Remove references from parent elements
        while elem.getprevious() is not None:
            del elem.getparent()[0]

The loop while elem.getprevious() is not None: is critical. It deletes finished sibling elements from the parent node. Without this, the parent node keeps pointers to every child element parsed since the beginning of the file.

Designing the ETL Pipeline

A production pipeline needs to do more than parse. It must clean data, batch it, and write it to a database.

Running these steps sequentially in a single thread creates bottlenecks. When the script parses XML, the database connection sits idle. When the script writes to the database, the parser waits.

We can solve this by separating the pipeline into three stages:

1. Reader (Parser): Reads the XML stream and extracts raw dictionaries.
2. Transformer: Normalizes data types, parses dates, and filters records.
3. Writer (Loader): Accumulates records and performs batch inserts.

Using Python's generator pattern allows data to flow through these stages without loading the entire dataset into memory. If running this asynchronously, you can manage the pipeline using designing fault-tolerant background queues in Postgres.

[XML File] -> [iterparse Generator] -> [Transform Generator] -> [Batch Accumulator] -> [DB Bulk Insert]

Implementing the Pipeline

Here is a complete implementation using Python, lxml, and PostgreSQL. We use psycopg2 to perform fast batch inserts.

First, set up the target database schema:

CREATE TABLE health_records (
    id SERIAL PRIMARY KEY,
    record_type VARCHAR(255) NOT NULL,
    source_name VARCHAR(255),
    value NUMERIC,
    unit VARCHAR(50),
    start_date TIMESTAMP WITH TIME ZONE NOT NULL,
    end_date TIMESTAMP WITH TIME ZONE NOT NULL
);
 
CREATE INDEX idx_records_type_date ON health_records(record_type, start_date);

Next, write the Python ETL script:

import sys
from datetime import datetime
from lxml import etree
import psycopg2
from psycopg2.extras import execute_values
 
DB_DSN = "dbname=health_db user=postgres password=secret host=localhost"
BATCH_SIZE = 10000
 
def stream_xml_elements(file_path):
    """Streams specific elements from XML file while keeping memory flat."""
    context = etree.iterparse(
        file_path, 
        events=('end',), 
        tag='Record', 
        huge_tree=True
    )
    for event, elem in context:
        # Extract attributes into a plain dictionary
        record_data = dict(elem.attrib)
        yield record_data
        
        # Free memory
        elem.clear()
        while elem.getprevious() is not None:
            del elem.getparent()[0]
 
def transform_record(raw_record):
    """Cleans and formats raw XML attributes."""
    try:
        # Convert date strings to datetime objects
        # Apple Health dates format: 'YYYY-MM-DD HH:MM:SS -HHMM'
        start_dt = datetime.strptime(raw_record['startDate'], '%Y-%m-%d %H:%M:%S %z')
        end_dt = datetime.strptime(raw_record['endDate'], '%Y-%m-%d %H:%M:%S %z')
        
        # Convert value to float if present
        raw_val = raw_record.get('value')
        val = float(raw_val) if raw_val and raw_val.replace('.', '', 1).isdigit() else None
        
        return (
            raw_record.get('type'),
            raw_record.get('sourceName'),
            val,
            raw_record.get('unit'),
            start_dt,
            end_dt
        )
    except (ValueError, KeyError):
        # Skip malformed records
        return None
 
def write_batches(connection, batch):
    """Inserts a batch of records using PostgreSQL COPY or execute_values."""
    query = """
        INSERT INTO health_records (record_type, source_name, value, unit, start_date, end_date)
        VALUES %s
    """
    with connection.cursor() as cursor:
        execute_values(cursor, query, batch)
    connection.commit()
 
def run_etl(file_path):
    conn = psycopg2.connect(DB_DSN)
    batch = []
    processed_count = 0
    
    print("Starting ETL pipeline...")
    
    for raw_rec in stream_xml_elements(file_path):
        transformed = transform_record(raw_rec)
        if not transformed:
            continue
            
        batch.append(transformed)
        
        if len(batch) >= BATCH_SIZE:
            write_batches(conn, batch)
            processed_count += len(batch)
            print(f"Processed {processed_count} records...")
            batch.clear()
            
    # Write remaining records
    if batch:
        write_batches(conn, batch)
        processed_count += len(batch)
        
    conn.close()
    print(f"ETL Complete. Total records imported: {processed_count}")
 
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python etl.py <path_to_export.xml>")
        sys.exit(1)
    run_etl(sys.argv[1])

Optimizing Database Ingestion

Inserting records one by one will kill performance. If you have 5 million records and each insert takes 2 milliseconds, the process will take nearly three hours.

We use three optimization techniques in the code above (for more scaling tips, see optimizing PostgreSQL query performance on large scale tables):

1. Batching

Instead of sending an insert statement for every record, we group them into batches of 10,000. This reduces network roundtrips between the script and the database.

2. execute_values

The psycopg2.extras.execute_values helper page-aligns the insert statements. It formats the query to insert multiple rows in a single command:

INSERT INTO health_records (...) VALUES (...), (...), (...);

This is significantly faster than executing INSERT inside a loop.

3. Disabling Indexes During Import

If you are importing into an empty database or performing a full reload, indexes can slow down the process. The database must update the index tree for every batch.

For maximum speed, drop the indexes before running the import, and recreate them afterward. For complex schemas, review multitenant database index tuning in PostgreSQL.

- Drop index
DROP INDEX IF EXISTS idx_records_type_date;
 
- Run ETL script here
 
- Recreate index
CREATE INDEX idx_records_type_date ON health_records(record_type, start_date);

Handling Schema Drift

Apple Health XML schemas change when iOS updates. New fields appear, and attribute names change.

If your database schema is rigid, the pipeline will fail when it encounters a new attribute. You can handle this by storing raw attributes in a PostgreSQL JSONB column.

CREATE TABLE health_records_flexible (
    id SERIAL PRIMARY KEY,
    record_type VARCHAR(255) NOT NULL,
    start_date TIMESTAMP WITH TIME ZONE NOT NULL,
    raw_data JSONB
);

Modify the transformer to extract the core fields and keep the rest as JSON:

import json
 
def transform_to_jsonb(raw_record):
    start_dt = datetime.strptime(raw_record['startDate'], '%Y-%m-%d %H:%M:%S %z')
    record_type = raw_record.pop('type')
    
    # The remaining keys in raw_record go into the JSON payload
    return (
        record_type,
        start_dt,
        json.dumps(raw_record)
    )

This approach prevents pipeline failures when Apple introduces new metadata fields. You can query the JSON data directly using PostgreSQL JSON operators:

SELECT raw_data->>'value' AS value 
FROM health_records_flexible 
WHERE record_type = 'HKQuantityTypeIdentifierStepCount';

Handling Interrupted Runs

If a 10 GB import fails halfway through, starting over from the beginning is inefficient. You need a way to resume.

Because XML is a hierarchical format, you cannot easily jump to a specific line without parsing the preceding tags. However, you can track progress by recording the byte offset of the file.

Python's file.tell() returns the current byte position. You can log this position to a state file periodically.

def stream_with_checkpoint(file_path, checkpoint_bytes=0):
    with open(file_path, 'rb') as f:
        if checkpoint_bytes:
            f.seek(checkpoint_bytes)
            
        context = etree.iterparse(f, events=('end',), tag='Record')
        for event, elem in context:
            current_offset = f.tell()
            yield elem, current_offset
            
            elem.clear()
            while elem.getprevious() is not None:
                del elem.getparent()[0]

If the process dies, read the last saved byte offset from your state database, seek to that position, and resume parsing.

Using these streaming techniques keeps memory usage under 100 MB, whether you are processing a 10 MB test file or a 10 GB production dataset.