CMS Medicare Streaming Data Pipeline

Submitted charge

The source's average covered hospital charge per DRG discharge; a billed amount, not an estimate of cost.

Medicare payment

The source's average Medicare payment measure. Total payment is retained separately and includes other payment components.

Charge ratio

Spark averages each input row's charge/payment ratio. A high ratio is an analytical signal, not hospital profit or confirmed misconduct.

dbt models

03

Declared dbt tests

09

Processing-time trigger

30s

The trigger is a configured processing cadence, not a measured end-to-end latency. CMS refreshes this public dataset annually; the project simulates streaming by sending retrieved records through Kafka.

• CMS Dataset Overview Opens in a new tab
• CMS Field Dictionary Opens in a new tab
• View Kafka Producer Opens in a new tab
• View Spark Streaming Job Opens in a new tab

• View Kafka Producer Opens in a new tab
• View Spark Streaming Job Opens in a new tab
• View Snowflake Writer Opens in a new tab
• View dbt Sources Opens in a new tab
• View Documented Workflow Opens in a new tab

1. 01 / RetrieveCMS API pages

   Use size and offset; request up to 1,000 rows per page, stopping at the 5,000-record target or an empty response.

2. 02 / SerializeUTF-8 JSON

   Publish each complete source row. DRG_Cd becomes the key; an absent field defaults to UNKNOWN.

3. 03 / Publishcms-inpatient-stream

   Send the page of messages and flush the producer. A half-second sleep between pages simulates arrival gaps.

4. 04 / ConsumeSpark Kafka source

   Read the topic from earliest offsets for a new query, then parse the payload into the selected schema.

BATCH_SIZE = 1000 # How many rows importing from API
TOTAL_ROWS = 5000 # Verify after first 5000 rows, full importing 145000 later on

        key = row.get("DRG_Cd", "UNKNOWN")
        producer.send(TOPIC, key=key, value=row)
    producer.flush()

Source GUID: 690ddc6c-2767-4618-b277-420ffb2bf27c. A read-only API audit on October 14, 2026 reported 145,879 available rows; catalog metadata identifies an annual release covering 2024. This source count is separate from the producer target and does not prove how many records reached Kafka.

• View Source Exploration Opens in a new tab
• CMS API Source Statistics Opens in a new tab
• CMS API Public Record Sample Opens in a new tab
• View Documented Workflow Opens in a new tab

hospital_count

COUNT(*)

Counts input messages in each DRG code/description group; not distinct provider CCNs.

avg_charge / avg_medicare_pymt / avg_total_pymt

ROUND(AVG(field), 2)

Unweighted averages of the provider/DRG summary records received by Spark.

avg_charge_ratio

ROUND(AVG(charge / medicare_payment), 2)

Average of individual source-row ratios, not AVG(charge) / AVG(payment). Discharges are cast but not used as weights.

            spark_round(
                avg(col("charge") / col("medicare_payment")), 2
            ).alias("avg_charge_ratio"),

        drg_agg.writeStream
        .outputMode("complete")
        .foreachBatch(write_to_snowflake)
        .trigger(processingTime="30 seconds")
        .start()

• View Spark Streaming Job Opens in a new tab
• View Kafka Connectivity Probe Opens in a new tab

        cursor.execute("TRUNCATE TABLE STREAMING.DRG_CHARGE_ANALYSIS")

        # ... INSERT statement and row tuples omitted ...

        cursor.executemany(insert_sql, data)
        conn.commit()

The snapshot table supplies updated_at through a Snowflake default, renamed batch_updated_at in staging. The mart preserves that snapshot timestamp; no historical version key or time-series accumulation is implemented.

• View Snowflake Table SQL Opens in a new tab
• View dbt Configuration Opens in a new tab
• View dbt Sources Opens in a new tab
• View Snowflake Writer Opens in a new tab

        CASE
            WHEN b.avg_charge_ratio > (s.mean_ratio + 2 * s.stddev_ratio)
            THEN TRUE
            ELSE FALSE
        END                             AS is_charge_outlier,

        CASE
            WHEN b.avg_charge_ratio >= 15 THEN 'Critical'
            WHEN b.avg_charge_ratio >= 10 THEN 'High'
            WHEN b.avg_charge_ratio >= 5 THEN 'Medium'
            ELSE 'Normal'
        END                             AS charge_severity

Nine tests are defined: five on staging and four on the mart. The README reports nine tests with zero failures, but no machine-readable result or execution log independently verifies that result. No fake pass indicator or sample financial table is reproduced.

• View Nine Test Declarations Opens in a new tab
• View Outlier & Severity SQL Opens in a new tab
• View Documented Workflow Opens in a new tab

1. 01 / Documented task

   run_cms_producer

   Retrieve API records and publish Kafka messages.

   Implementation boundary

   Producer code is present; the Airflow invocation is documented only.

2. 02 / Documented task

   run_spark_streaming

   Start Spark processing and Snowflake snapshot writes.

   Implementation boundary

   The job awaits termination indefinitely; the DAG's stop/completion behavior is not published.

3. 03 / Documented task

   run_dbt_models

   Build the staging/intermediate views and mart table.

   Implementation boundary

   The documented dependency requires the preceding task to complete.

4. 04 / Documented task

   run_dbt_tests

   Run the nine declared model checks.

   Implementation boundary

   No task logs or test execution artifact establish successful completion.

• View Airflow Compose Opens in a new tab
• View Airflow Dockerfile Opens in a new tab
• View Spark Streaming Job Opens in a new tab
• View Documented Workflow Opens in a new tab

Unverified sample financial results and the README graphic’s full-run/pass labels were omitted. No patient data, source account identifiers, passwords, or private endpoints were copied. The snapshot sink does not establish exactly-once writes, complete history, observed availability, measured latency, or production operation.

Source reviewed at 5143bdc. The Kafka partition figure is illustrative; the other figures explain source-defined code or clearly labeled documented workflows. Charge severity and statistical outliers require business context before any operational decision.

• View Documented Workflow Opens in a new tab
• View Kafka Producer Opens in a new tab
• View Spark Streaming Job Opens in a new tab
• View Snowflake Writer Opens in a new tab
• View dbt Configuration Opens in a new tab
• View Nine Test Declarations Opens in a new tab