Distributed FHIR Processing with Apache Spark

When a FHIR Bulk Data export lands as tens of gigabytes of NDJSON, a single-process pandas parser stops being viable and the work has to move onto a cluster. PySpark reads that NDJSON natively — one line is one FHIR resource — but the deeply nested, repeated structure of resources like Observation and Patient makes naive schema inference slow, unstable, and non-deterministic across files. This page is the Spark-native companion to asynchronous batch processing for large clinical datasets: it shows how to pin an explicit StructType, explode repeated elements, flatten nested structs into a columnar analytics table, and write skew-resistant partitioned Parquet without leaking PHI through the shuffle.

FHIR shape to Spark handling (quick reference)

FHIR R4 resources are recursive JSON: scalars sit beside nested objects (BackboneElement), and almost every clinically interesting field is a repeated element or a CodeableConcept. Spark can represent all of it, but each shape needs a specific handling idiom. Keep this table next to your transform — it is the mapping from a FHIR construct to the exact column expression that flattens it.

Two rules drive everything below. First, dot access reaches into a struct without exploding — it is cheap and keeps row count flat. Second, explode() is the only correct way to fan an array into rows, and it multiplies row count, so every explode must be a deliberate, tested decision, not an accident of a wide select.

Implementation pattern

Pin an explicit schema — never infer

spark.read.json will infer a schema by scanning the input, but over FHIR this is a trap. Inference reads a sampling (or all) of the data before any real work starts, which is slow at bulk-export scale; worse, it is unstable. If file A never contains Observation.component and file B does, the inferred struct differs between runs, and a field absent from the sample is silently dropped — your component column simply will not exist. An explicit StructType fixes the contract: fields you declare are read, fields you omit are ignored, and the job is deterministic across every file in the export.

from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, ArrayType,
)

spark = (
    SparkSession.builder
    .appName("fhir-observation-flatten")
    # Encrypt shuffle traffic and on-disk spill — PHI leaves memory here.
    .config("spark.io.encryption.enabled", "true")
    .config("spark.network.crypto.enabled", "true")
    .config("spark.local.dir", "/mnt/encrypted-scratch")
    .getOrCreate()
)

# Reusable CodeableConcept.coding shape: system + code + display.
coding = ArrayType(StructType([
    StructField("system", StringType()),
    StructField("code", StringType()),
    StructField("display", StringType()),
]))

# Explicit Observation schema. Declare only the elements the analytics
# table needs; everything else in the NDJSON is skipped, not inferred.
observation_schema = StructType([
    StructField("resourceType", StringType()),
    StructField("id", StringType()),
    StructField("status", StringType()),
    StructField("effectiveDateTime", StringType()),
    StructField("subject", StructType([
        StructField("reference", StringType()),
    ])),
    StructField("code", StructType([
        StructField("coding", coding),
    ])),
    # value[x] is polymorphic; model the quantity variant explicitly.
    StructField("valueQuantity", StructType([
        StructField("value", StringType()),
        StructField("unit", StringType()),
    ])),
    # Repeated BackboneElement — each entry has its own code + value.
    StructField("component", ArrayType(StructType([
        StructField("code", StructType([StructField("coding", coding)])),
        StructField("valueQuantity", StructType([
            StructField("value", StringType()),
            StructField("unit", StringType()),
        ])),
    ]))),
])

Modeling effectiveDateTime and valueQuantity.value as StringType is deliberate: FHIR dates carry variable precision (2026, 2026-07, 2026-07-15T09:30:00Z) and quantities can be non-numeric, so read them as text and cast downstream where you can quarantine bad values rather than letting Spark null them silently.

Read, flatten, and explode

With the schema pinned, the read is a one-liner and the rest is column expressions. Dot access flattens the nested structs; explode fans component into one row per sub-observation. This mirrors the columnar flattening covered for the in-memory path in optimizing pandas for FHIR JSON parsing — same target shape, distributed engine.

from pyspark.sql import functions as F

raw = (
    spark.read
    .schema(observation_schema)   # explicit — no inference pass
    .json("s3://phi-bulk-export/Observation/*.ndjson")
)

# Flatten the top-level observation into analytic columns.
flat = raw.select(
    F.col("id").alias("observation_id"),
    F.col("status"),
    F.col("effectiveDateTime").alias("effective"),
    # Reference "Patient/123" -> patient_id "123" for joins.
    F.element_at(F.split("subject.reference", "/"), -1).alias("patient_id"),
    # Primary code: pick the first coding of the CodeableConcept.
    F.col("code.coding")[0]["system"].alias("code_system"),
    F.col("code.coding")[0]["code"].alias("code"),
    F.col("valueQuantity.value").alias("value"),
    F.col("valueQuantity.unit").alias("unit"),
    F.col("component"),   # keep the array for the explode below
)

# Explode repeated components: one row per (observation, component).
# explode_outer keeps observations that have no component (single-value obs).
components = (
    flat
    .select(
        "observation_id", "patient_id", "effective",
        F.explode_outer("component").alias("comp"),
    )
    .select(
        "observation_id", "patient_id", "effective",
        F.col("comp.code.coding")[0]["code"].alias("component_code"),
        F.col("comp.valueQuantity.value").alias("component_value"),
        F.col("comp.valueQuantity.unit").alias("component_unit"),
    )
)

Choosing explode_outer over explode matters: plain explode drops any observation whose component array is null or empty, so a blood-pressure panel (which uses components) survives but a lone glucose reading (which does not) silently vanishes from the output. For a panel-only analytics table use explode; for a complete per-resource ledger use explode_outer.

Partition the output for downstream queries

Writing Parquet partitioned by resourceType and an event date turns every downstream query into a partition prune — a dashboard asking for one day of Observation reads one directory instead of the whole export. Derive the partition date from effective, and be explicit about repartitioning before the write to control file count.

out = components.withColumn(
    "date", F.to_date(F.col("effective"))
).withColumn("resourceType", F.lit("Observation"))

(
    out
    # Co-locate rows by partition key before writing to avoid many
    # tiny files (one per input task per partition).
    .repartition("resourceType", "date")
    .write
    .partitionBy("resourceType", "date")
    .mode("overwrite")
    .parquet("s3://phi-analytics/fhir/observation/")
)

The repartition before partitionBy is the small-files fix: without it, every input task writes its own fragment into every date directory, producing thousands of sub-megabyte Parquet files that cripple later reads. Feed the partitioned output straight into a warehouse model — the dimensional layer built in modeling clinical data with dbt consumes exactly this resourceType/date layout.

Validation and testing

Exploded output is easy to get subtly wrong — an off-by-one on row count usually means the wrong explode variant. Test the transform against a tiny fixture whose expected row count you can compute by hand, and assert that primary keys are never null (a null observation_id breaks every downstream join).

def flatten_observations(df):
    """The transform under test — same logic as the pipeline above."""
    return (
        df.select(
            F.col("id").alias("observation_id"),
            F.explode_outer("component").alias("comp"),
        )
    )

def test_explode_row_count(spark):
    # Fixture: 2 observations, one with 2 components, one with none.
    # explode_outer => 2 rows for the panel + 1 null-comp row = 3.
    df = spark.read.schema(observation_schema).json("fixtures/obs_small.ndjson")
    result = flatten_observations(df)
    assert result.count() == 3, result.count()

def test_no_null_primary_key(spark):
    df = spark.read.schema(observation_schema).json("fixtures/obs_small.ndjson")
    result = flatten_observations(df)
    # No exploded row may lose its observation_id.
    assert result.filter(F.col("observation_id").isNull()).count() == 0

Run these against a local SparkSession in CI. Freeze the fixture NDJSON alongside its expected counts so that a change in explode semantics — or a schema field that starts arriving as an array instead of a scalar — fails the assertion instead of silently multiplying or dropping rows in production. The same page-through-the-source discipline that produces these fixtures is covered in paginating FHIR Bulk Export NDJSON in Python.

Gotchas and compliance constraints

• Schema inference over FHIR is the headline trap. Letting spark.read.json infer the schema costs a full pre-scan and produces a struct that changes with the sample — a component present in one export and absent in the next yields different columns on different runs, breaking downstream models without a code change. Pin an explicit StructType, declare only what you consume, and version that schema alongside your code.
• Data skew on high-volume patients stalls the job. A handful of patients with years of monitoring data can hold millions of observations; any groupBy or partitionBy keyed on patient_id sends them all to one executor, and that task runs long after every other finishes. Detect it by inspecting partition sizes, then salt the hot key (append a random bucket suffix, aggregate, then re-combine) or enable adaptive query execution with spark.sql.adaptive.skewJoin.enabled so Spark splits the skewed partition automatically.
• PHI travels through shuffle and spill — encrypt both. Every explode, repartition, and wide transform moves patient data across the network and, under memory pressure, spills it to local disk in cleartext by default. Set spark.io.encryption.enabled and spark.network.crypto.enabled, point spark.local.dir at an encrypted volume, and treat executor scratch space as an in-scope PHI store under the HIPAA Security Rule. Strip or hash direct identifiers before they reach any shared analytics table.

Deployment checklist

• Explicit schema pinned — StructType declared and version-controlled; no inferSchema path in production.
• Explode variant chosen deliberately — explode vs explode_outer matches whether empty-array resources must survive.
• Small-files guard — repartition on the partition keys before partitionBy write.
• Skew mitigation enabled — adaptive skew handling on, or hot patient keys salted.
• Shuffle and spill encrypted — IO/network crypto on, spark.local.dir on an encrypted volume.
• PHI minimized — direct identifiers dropped or hashed before the analytics write.

Treated this way, PySpark turns a multi-gigabyte Bulk Data export into a deterministic, partition-pruned analytics table without ever trusting inference or leaking PHI through the cluster fabric.

Related

• Asynchronous batch processing for large clinical datasets — parent guide: the batch patterns this Spark job plugs into
• Modeling clinical data with dbt — the warehouse layer that consumes the partitioned Parquet output
• Paginating FHIR Bulk Export NDJSON in Python — how the NDJSON that Spark reads is fetched from the source
• Optimizing pandas for FHIR JSON parsing — the single-node flattening path for datasets that fit in memory