agentsclimarketplace

Spark healthcare data pipeline

Skill rbr7/MedClawMini/skills/spark-healthcare-data-pipeline

A focused, production-minded library of 197 clinical-AI and healthcare data-science skills for the OpenClaw agent platform featuring data quality, clinical NLP, big-data ML, explainable AI, drug safety, and regulatory.

Install
npx -y skills add rbr7/MedClawMini --skill spark-healthcare-data-pipeline

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

One thing to look at

  • 0 stars0 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

What its author says it does

Copied from the file, not written here

Build scalable, production-grade data pipelines for large healthcare datasets (billions of claims/eligibility/EHR rows) using Apache Spark on Hadoop/Delta Lake. Covers PySpark and Scala Spark, partitioning and bucketing, joins and skew handling, window-function feature engineering, schema enforcement, incremental/idempotent batch jobs, and performance tuning (broadcast joins, caching, AQE). Use when data volume exceeds a single machine, or to productionize claims ETL and ML feature pipelines at scale.

The file declares its own license as MIT. That is the author’s claim about this one file, and it is not the same thing as the license GitHub reports for the repository, which is listed with the other numbers below.

SKILL.md

4.9 KB, as published. Nobody here has run it

Spark Healthcare Data Pipeline

Overview

When a claims feed is billions of rows, pandas stops and distributed compute begins. This skill builds Apache Spark pipelines that ingest, validate, transform, and feature- engineer healthcare data at scale, with the production concerns that matter: schema enforcement, partitioning, skew, idempotency, and cost. It covers both PySpark and Scala Spark (Scala being the JVM-native, type-safe path many platforms standardize on).

When to Use This Skill

  • Data volume exceeds single-machine memory (claims, eligibility, Rx, EHR at population scale).
  • Productionizing the rules from healthcare-data-quality-profiling or the matching from patient-record-entity-resolution over the full dataset.
  • Building reusable ML feature tables with windowed aggregations (e.g., 6-month utilization per member).
  • Standing up incremental, idempotent batch jobs on Hadoop/Delta/cloud object storage.

Engineering Practices

  1. Schema first enforce an explicit StructType; reject/quarantine off-schema rows rather than silently coercing.
  2. Partitioning partition by a high-selectivity, time-based key (e.g., service_year_month); avoid tiny-file and skew traps; bucket large join keys.
  3. Joins broadcast small dimensions (provider, code lookups); salt skewed keys; enable Adaptive Query Execution (AQE).
  4. Feature engineering window functions for per-member rolling counts/costs; pivot diagnosis/procedure histories; everything reproducible and point-in-time correct (no label leakage).
  5. Idempotency MERGE/upsert into Delta so reruns are safe; checkpoint and track watermarks for incremental loads.
  6. Tuning & cost cache hot DataFrames, right-size partitions (spark.sql.shuffle.partitions), monitor the Spark UI for spills and skew.

Example

# PySpark: point-in-time member feature table from claims
from pyspark.sql import functions as F, Window
spark = SparkSession.builder.appName("member_features").getOrCreate()

claims = (spark.read.format("delta").load("/lake/claims")
          .filter(F.col("service_date").isNotNull()))

w6 = (Window.partitionBy("member_id").orderBy(F.col("service_date").cast("long"))
      .rangeBetween(-180*86400, 0))               # trailing 180 days

features = (claims
    .withColumn("visits_6m", F.count("claim_id").over(w6))
    .withColumn("cost_6m",   F.sum("paid_amount").over(w6))
    .withColumn("ed_6m", F.sum(F.when(F.col("place_of_service")=="23",1).otherwise(0)).over(w6))
    .groupBy("member_id", "as_of_month")
    .agg(F.max("visits_6m").alias("visits_6m"),
         F.max("cost_6m").alias("cost_6m"),
         F.max("ed_6m").alias("ed_visits_6m")))

(features.repartition("as_of_month")
    .write.format("delta").mode("overwrite")
    .partitionBy("as_of_month").save("/lake/features/member_monthly"))

// Scala Spark: same shape, JVM-native and type-checked
val w6 = Window.partitionBy("member_id").orderBy($"service_date".cast("long"))
              .rangeBetween(-180*86400, 0)
val features = claims
  .withColumn("visits_6m", count("claim_id").over(w6))
  .withColumn("cost_6m",   sum("paid_amount").over(w6))
features.write.format("delta").mode("overwrite")
        .partitionBy("as_of_month").save("/lake/features/member_monthly")

Outputs

  • Partitioned Delta/Parquet feature and curated tables.
  • A parameterized, idempotent Spark job (PySpark or Scala) with schema contract.
  • pipeline_run.md row counts, reject/quarantine stats, runtime, partition layout.

Healthcare Context

Built for population-scale payer/provider data and the leakage-sensitive, point-in-time nature of healthcare ML. It is the scale-out backbone under healthcare-data-quality-profiling, patient-record-entity-resolution, and healthcare-predictive-modeling. Scala examples are included because production data platforms frequently standardize on JVM Spark for type safety and performance.

References

Keep looking

Skills are one crate of 328,083. Ordering is by how many stacks a row turns up in, so the top of any crate is what has actually been picked rather than what has the most stars.