Parsing HL7 Telemetry for Healthcare Instruments

Processing raw, unstructured HL7 medical telemetry batches in PySpark or lightweight native Python to compute vital signs statistics and alert anomalies.

The Setup

Medical telemetry devices (patient monitors, ECGs, blood gas analyzers) in intensive care units stream telemetry to a local collector. The collector aggregates these text files in 10-minute batch windows on S3/HDFS.

The messages use the HL7 v2 (Health Level Seven) industry standard, which is a pipe-delimited (|) format where segments are separated by carriage returns (\r or \n).

Sample Raw HL7 Input file (10-minute batch)

Each message begins with an MSH header, followed by patient (PID), request (OBR), and one or more observation (OBX) segments:

MSH|^~\&|MONITOR_ICU|BED_104|EMR_GATEWAY|HOSPITAL_A|20260531180000||ORU^R01|MSG_001|P|2.5
PID|1||PAT_88329||SMITH^JOHN||19780415|M
OBR|1|||HEART_MONITOR|||20260531180000
OBX|1|NM|88-5^HEART_RATE||82|bpm|60-100|N|||F
OBX|2|NM|100-3^BLOOD_PRESSURE_SYS||135|mmHg|90-120|H|||F
OBX|3|NM|100-4^BLOOD_PRESSURE_DIA||85|mmHg|60-80|H|||F
---
MSH|^~\&|MONITOR_ICU|BED_108|EMR_GATEWAY|HOSPITAL_A|20260531180500||ORU^R01|MSG_002|P|2.5
PID|1||PAT_44120||DOE^JANE||19850923|F
OBR|1|||HEART_MONITOR|||20260531180500
OBX|1|NM|88-5^HEART_RATE||115|bpm|60-100|H|||F
OBX|2|NM|100-3^BLOOD_PRESSURE_SYS||110|mmHg|90-120|N|||F
OBX|3|NM|100-4^BLOOD_PRESSURE_DIA||70|mmHg|60-80|N|||F

Use Case 1: High-Volume 10-Minute Batches (PySpark Pipeline)

When handling large hospitals with thousands of active beds streaming telemetry continuously over 10-minute intervals, PySpark distributed compute handles the scalability.

Step 1: Read Whole Text Batch & Isolate Messages

We read the raw telemetry file in a single block using wholetext = true, then split messages by the separator (---) and explode them into separate records.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("HL7_Telemetry_Parser").getOrCreate()

# Read the raw text batch as a single block
raw_text_df = spark.read \
    .option("wholetext", "true") \
    .text("s3://healthcare-lake/telemetry-batches/")

# Split and explode by message separator
messages_df = raw_text_df.select(
    F.explode(F.split(F.col("value"), "---")).alias("raw_message")
).filter(F.trim(F.col("raw_message")) != "")

messages_df.show(truncate=60)

Intermediate DataFrame 1 (messages_df):

+------------------------------------------------------------+
|                                                 raw_message|
+------------------------------------------------------------+
|MSH|^~\&|MONITOR_ICU|BED_104|EMR_GATEWAY|HOSPITAL_A|202605...|
|MSH|^~\&|MONITOR_ICU|BED_108|EMR_GATEWAY|HOSPITAL_A|202605...|
+------------------------------------------------------------+

Step 2: Split Messages into Segments

We split the raw message string by line break characters (\r or \n) into a structured list of individual segment strings.

messages_split_df = messages_df.withColumn(
    "segments", F.split(F.trim(F.col("raw_message")), "[\r\n]+")
)

messages_split_df.select("segments").show(truncate=80)

Intermediate DataFrame 2 (messages_split_df):

+--------------------------------------------------------------------------------+
|                                                                        segments|
+--------------------------------------------------------------------------------+
|[MSH|^~\&|MONITOR_ICU|BED_104|EMR_GATEWAY|HOSPITAL_A|20260531180000||ORU^R01...|
|[MSH|^~\&|MONITOR_ICU|BED_108|EMR_GATEWAY|HOSPITAL_A|20260531180500||ORU^R01...|
+--------------------------------------------------------------------------------+

Step 3: Extract Specific Positional Segments

We isolate the relevant HL7 segment lines (MSH, PID, and all OBX occurrences) using Spark array filtering functions.

parsed_segments_df = messages_split_df.select(
    F.filter(F.col("segments"), lambda x: x.startswith("MSH")).getItem(0).alias("msh_segment"),
    F.filter(F.col("segments"), lambda x: x.startswith("PID")).getItem(0).alias("pid_segment"),
    F.filter(F.col("segments"), lambda x: x.startswith("OBX")).alias("obx_segments")
)

parsed_segments_df.show(truncate=40)

Intermediate DataFrame 3 (parsed_segments_df):

+----------------------------------------+----------------------------------------+----------------------------------------+
|                             msh_segment|                             pid_segment|                            obx_segments|
+----------------------------------------+----------------------------------------+----------------------------------------+
|MSH|^~\&|MONITOR_ICU|BED_104|EMR_GAT...|PID|1||PAT_88329||SMITH^JOHN||1978...|[OBX|1|NM|88-5^HEART_RATE||82|bpm...|
|MSH|^~\&|MONITOR_ICU|BED_108|EMR_GAT...|PID|1||PAT_44120||DOE^JANE||19850...|[OBX|1|NM|88-5^HEART_RATE||115|bp...|
+----------------------------------------+----------------------------------------+----------------------------------------+

Step 4: Parse MSH Positional Fields

HL7 is positional. We split the isolated MSH segment by the standard pipe character (|) to extract instrument ID, bed location, and transmission time.

msh_parsed = parsed_segments_df.withColumn(
    "msh_fields", F.split(F.col("msh_segment"), "\\|")
).select(
    "*",
    F.col("msh_fields").getItem(3).alias("device_id"),
    F.col("msh_fields").getItem(4).alias("location_bed"),
    F.to_timestamp(F.col("msh_fields").getItem(6), "yyyyMMddHHmmss").alias("message_time")
).drop("msh_fields", "msh_segment")

msh_parsed.select("device_id", "location_bed", "message_time").show()

Intermediate DataFrame 4 (msh_parsed):

+-----------+------------+-------------------+
|  device_id|location_bed|       message_time|
+-----------+------------+-------------------+
|MONITOR_ICU|     BED_104|2026-05-31 18:00:00|
|MONITOR_ICU|     BED_108|2026-05-31 18:05:00|
+-----------+------------+-------------------+

Step 5: Parse PID Patient Fields

Next, we split the PID segment by pipe (|), extracting patient ID, birth date, gender, and splitting patient name components by caret (^).

pid_parsed = msh_parsed.withColumn(
    "pid_fields", F.split(F.col("pid_segment"), "\\|")
).select(
    "*",
    F.col("pid_fields").getItem(3).alias("patient_id"),
    F.split(F.col("pid_fields").getItem(5), "\\^").getItem(0).alias("last_name"),
    F.split(F.col("pid_fields").getItem(5), "\\^").getItem(1).alias("first_name"),
    F.col("pid_fields").getItem(7).alias("birth_date"),
    F.col("pid_fields").getItem(8).alias("gender")
).drop("pid_fields", "pid_segment")

pid_parsed.select("patient_id", "last_name", "first_name", "gender").show()

Intermediate DataFrame 5 (pid_parsed):

+----------+---------+----------+------+
|patient_id|last_name|first_name|gender|
+----------+---------+----------+------+
| PAT_88329|    SMITH|      JOHN|     M|
| PAT_44120|      DOE|      JANE|     F|
+----------+---------+----------+------+

Step 6: Explode & Positional Parse Multiple OBX Segments

Since each message contains multiple observation lines (obx_segments array), we explode this column to create a separate row for each clinical metric. We then parse the individual observation fields.

# Explode multiple observations to individual rows
exploded_obx = pid_parsed.select(
    "patient_id", "last_name", "first_name", "gender", "device_id", "location_bed", "message_time",
    F.explode(F.col("obx_segments")).alias("single_obx")
)

# Extract and cast individual clinical telemetry metrics
structured_telemetry = exploded_obx.withColumn(
    "obx_fields", F.split(F.col("single_obx"), "\\|")
).select(
    "patient_id",
    F.concat_ws(", ", F.col("last_name"), F.col("first_name")).alias("patient_name"),
    "gender",
    "device_id",
    "location_bed",
    "message_time",
    F.split(F.col("obx_fields").getItem(3), "\\^").getItem(1).alias("metric_name"),
    F.col("obx_fields").getItem(5).cast("double").alias("metric_value"),
    F.col("obx_fields").getItem(6).alias("metric_unit"),
    F.col("obx_fields").getItem(7).alias("normal_reference"),
    F.col("obx_fields").getItem(8).alias("abnormal_flag")
)

# Cache structured dataset for analytical Spark SQL queries
structured_telemetry.cache()
structured_telemetry.createOrReplaceTempView("patient_telemetry")

structured_telemetry.show(truncate=False)

Final Structured DataFrame (structured_telemetry):

+----------+------------+------+-----------+------------+-------------------+---------------------+------------+-----------+----------------+-------------+
|patient_id|patient_name|gender|device_id  |location_bed|message_time       |metric_name          |metric_value|metric_unit|normal_reference|abnormal_flag|
+----------+------------+------+-----------+------------+-------------------+---------------------+------------+-----------+----------------+-------------+
|PAT_88329 |SMITH, JOHN |M      |MONITOR_ICU|BED_104     |2026-05-31 18:00:00|HEART_RATE           |82.0        |bpm        |60-100          |N            |
|PAT_88329 |SMITH, JOHN |M      |MONITOR_ICU|BED_104     |2026-05-31 18:00:00|BLOOD_PRESSURE_SYS   |135.0       |mmHg       |90-120          |H            |
|PAT_88329 |SMITH, JOHN |M      |MONITOR_ICU|BED_104     |2026-05-31 18:00:00|BLOOD_PRESSURE_DIA   |85.0        |mmHg       |60-80           |H            |
|PAT_44120 |DOE, JANE   |F      |MONITOR_ICU|BED_108     |2026-05-31 18:05:00|HEART_RATE           |115.0       |bpm        |60-100          |H            |
|PAT_44120 |DOE, JANE   |F      |MONITOR_ICU|BED_108     |2026-05-31 18:05:00|BLOOD_PRESSURE_SYS   |110.0       |mmHg       |90-120          |N            |
|PAT_44120 |DOE, JANE   |F      |MONITOR_ICU|BED_108     |2026-05-31 18:05:00|BLOOD_PRESSURE_DIA   |70.0        |mmHg       |60-80           |N            |
+----------+------------+------+-----------+------------+-------------------+---------------------+------------+-----------+----------------+-------------+

Spark SQL Clinical Statistics

-- Count Patients with Critical or Abnormal Flags
SELECT patient_id, patient_name, location_bed, metric_name, metric_value, abnormal_flag
FROM patient_telemetry
WHERE abnormal_flag IN ('H', 'L');

-- General ICU Statistics
SELECT metric_name, metric_unit, COUNT(DISTINCT patient_id) as patients, ROUND(AVG(metric_value), 2) as avg_val
FROM patient_telemetry
GROUP BY metric_name, metric_unit;

Use Case 2: Lightweight 2-3 Minute Batches (Pure Python Pipeline)

When telemetry batches arrive frequently (every 2-3 minutes), the data volume is highly compact. Triggering a JVM-heavy PySpark cluster introduces massive overhead compared to execution time.

For these frequent, smaller intervals, a pure Python single-node processing script using standard libraries is significantly more cost-effective and faster.

Pure Python Positional HL7 Parser

Here is the complete lightweight script using Python's standard library to read, parse, and analyze 2-3 minute medical telemetry files:

import re
from datetime import datetime
from collections import defaultdict

def parse_hl7_batch(file_content):
    """
    Reads the raw HL7 file content, splits messages by separator,
    and returns a structured list of patient observation dicts.
    """
    structured_records = []
    # Split raw batch text into individual messages
    messages = [msg.strip() for msg in file_content.split("---") if msg.strip()]

    for raw_message in messages:
        # Split message into segments by carriage returns or newlines
        segments = [seg.strip() for seg in re.split(r'[\r\n]+', raw_message) if seg.strip()]

        # Isolate segments
        msh_line = next((s for s in segments if s.startswith("MSH")), None)
        pid_line = next((s for s in segments if s.startswith("PID")), None)
        obx_lines = [s for s in segments if s.startswith("OBX")]

        if not msh_line or not pid_line:
            continue

        # 1. Parse MSH (Message Header)
        msh_fields = msh_line.split("|")
        device_id = msh_fields[3]
        location_bed = msh_fields[4]
        timestamp_str = msh_fields[6]
        # Normalize date string to Python datetime
        message_time = datetime.strptime(timestamp_str, "%Y%m%d%H%M%S")

        # 2. Parse PID (Patient Identification)
        pid_fields = pid_line.split("|")
        patient_id = pid_fields[3]
        name_fields = pid_fields[5].split("^")
        last_name = name_fields[0]
        first_name = name_fields[1] if len(name_fields) > 1 else ""
        patient_name = f"{last_name}, {first_name}"
        gender = pid_fields[8]

        # 3. Parse OBX (Observation Results)
        for obx_line in obx_lines:
            obx_fields = obx_line.split("|")
            metric_code_name = obx_fields[3].split("^")
            metric_name = metric_code_name[1] if len(metric_code_name) > 1 else "UNKNOWN"

            metric_value = float(obx_fields[5]) if obx_fields[5] else None
            metric_unit = obx_fields[6]
            normal_reference = obx_fields[7]
            abnormal_flag = obx_fields[8]

            structured_records.append({
                "patient_id": patient_id,
                "patient_name": patient_name,
                "gender": gender,
                "device_id": device_id,
                "location_bed": location_bed,
                "message_time": message_time,
                "metric_name": metric_name,
                "metric_value": metric_value,
                "metric_unit": metric_unit,
                "normal_reference": normal_reference,
                "abnormal_flag": abnormal_flag
            })

    return structured_records

# ==========================================
# Run the Lightweight Python Pipeline
# ==========================================

# Simulating reading raw batch file
raw_telemetry_batch = """
MSH|^~\&|MONITOR_ICU|BED_104|EMR_GATEWAY|HOSPITAL_A|20260531180000||ORU^R01|MSG_001|P|2.5
PID|1||PAT_88329||SMITH^JOHN||19780415|M
OBR|1|||HEART_MONITOR|||20260531180000
OBX|1|NM|88-5^HEART_RATE||82|bpm|60-100|N|||F
OBX|2|NM|100-3^BLOOD_PRESSURE_SYS||135|mmHg|90-120|H|||F
OBX|3|NM|100-4^BLOOD_PRESSURE_DIA||85|mmHg|60-80|H|||F
---
MSH|^~\&|MONITOR_ICU|BED_108|EMR_GATEWAY|HOSPITAL_A|20260531180500||ORU^R01|MSG_002|P|2.5
PID|1||PAT_44120||DOE^JANE||19850923|F
OBR|1|||HEART_MONITOR|||20260531180500
OBX|1|NM|88-5^HEART_RATE||115|bpm|60-100|H|||F
OBX|2|NM|100-3^BLOOD_PRESSURE_SYS||110|mmHg|90-120|N|||F
OBX|3|NM|100-4^BLOOD_PRESSURE_DIA||70|mmHg|60-80|N|||F
"""

# Parse Telemetry
records = parse_hl7_batch(raw_telemetry_batch)

Python-native Telemetry Statistics & Anomalies

With the data parsed into a standard list of Python dictionaries, we can calculate clinical statistics instantly using basic Python constructs.

1. Identify Patients with Abnormal Flags

print("--- CLINICAL ALERTS (ABNORMAL VITAL SIGNS) ---")
abnormal_records = [r for r in records if r["abnormal_flag"] in ("H", "L")]

for r in abnormal_records:
    print(f"ALERT: Patient {r['patient_name']} ({r['location_bed']}) "
          f"has abnormal {r['metric_name']}: {r['metric_value']} {r['metric_unit']} "
          f"({r['abnormal_flag']})")

Output:

--- CLINICAL ALERTS (ABNORMAL VITAL SIGNS) ---
ALERT: Patient SMITH, JOHN (BED_104) has abnormal BLOOD_PRESSURE_SYS: 135.0 mmHg (H)
ALERT: Patient SMITH, JOHN (BED_104) has abnormal BLOOD_PRESSURE_DIA: 85.0 mmHg (H)
ALERT: Patient DOE, JANE (BED_108) has abnormal HEART_RATE: 115.0 bpm (H)

2. General ICU Statistics Calculation

print("\n--- GENERAL ICU METRIC WARD STATISTICS ---")
metric_groups = defaultdict(list)
for r in records:
    metric_groups[(r["metric_name"], r["metric_unit"])].append(r["metric_value"])

for (name, unit), values in metric_groups.items():
    avg_val = sum(values) / len(values)
    min_val = min(values)
    max_val = max(values)
    print(f"Metric: {name:<20} | Avg: {avg_val:>6.2f} {unit} | Min: {min_val:>5.1f} | Max: {max_val:>5.1f} | Monitored: {len(values)} records")

Output:

--- GENERAL ICU METRIC WARD STATISTICS ---
Metric: HEART_RATE           | Avg:  98.50 bpm | Min:  82.0 | Max: 115.0 | Monitored: 2 records
Metric: BLOOD_PRESSURE_SYS   | Avg: 122.50 mmHg | Min: 110.0 | Max: 135.0 | Monitored: 2 records
Metric: BLOOD_PRESSURE_DIA   | Avg:  77.50 mmHg | Min:  70.0 | Max:  85.0 | Monitored: 2 records

Architectural Comparison Matrix

Real-World Sub-Scenarios

Sub-Scenario A: Message Parsing Failure and Delimiter Variations

Problem: A vendor update configures a blood analyzer to use standard Windows line endings (\r\n) instead of Linux (\n), causing the segments array to contain empty string items and throwing IndexOutOfBounds exceptions during positional parsing. Solution: Define an explicit regex pattern [\r\n]+ in the string splitting stage to treat consecutive line breaks as a single delimiter, and perform empty/null checks on elements before selecting positions.

Sub-Scenario B: Multi-Version Delimiter Compatibility (2.3 vs 2.5)

Problem: Older instruments send HL7 v2.3 messages without the full sub-component detail structure, causing caret (^) indexing to return null values. Solution: Use PySpark coalesce (or standard try-except/fallback checks in Python) to check multiple split fields and fallback to a default label (e.g. 'UNKNOWN_METRIC') if the caret component structure is missing.