Skip to content

convert_to_subject_sharded

Re-shard raw data by subject. See module docstring for details.

All arguments come through the Hydra cfg object; this stage has no stage-specific options beyond the global MESSY_config_fp.

Details

Property Value
Type main
Metadata stage False

Usage

MEDS_transform-stage <pipeline.yaml> convert_to_subject_sharded input_dir=<input> output_dir=<output>

Examples

join

Demonstrates _table.join — the vitals table has no subject_id column of its own, so _table.join.stays.key: stay_id, cols: [subject_id] brings in the subject_id from the stays table via a left join on stay_id. The stage materializes the joined frame, filters to the split’s subjects, and writes per-subject shards.

This example uses the stage’s config.yaml file.

Input files:

data/vitals/[0-3).parquet:
  stay_id: [10, 10, 20]
  charttime:
    ["01/01/2021 00:00:00", "01/01/2021 01:00:00", "01/01/2021 02:00:00"]
  HR: [70, 75, 65]

data/stays/[0-2).parquet:
  stay_id: [10, 20]
  subject_id: [111, 222]

metadata/.shards.json:
  train/0: [111]
  tuning/0: [222]

messy.yaml: |
  vitals:
    _table:
      join:
        stays:
          key: stay_id
          cols: [subject_id]
    HR:
      code: HR
      time: '$charttime::"%m/%d/%Y %H:%M:%S"'
      numeric_value: $HR

Expected output files:

data/train/0/vitals.parquet:
  stay_id: [10, 10]
  charttime: ["01/01/2021 00:00:00", "01/01/2021 01:00:00"]
  HR: [70, 75]
  subject_id: [111, 111]

data/tuning/0/vitals.parquet:
  stay_id: [20]
  charttime: ["01/01/2021 02:00:00"]
  HR: [65]
  subject_id: [222]

Run this stage:

MEDS_transform-stage <pipeline.yaml> convert_to_subject_sharded input_dir=<input> output_dir=<output>

default

Regroups the normalized source tables (from convert_to_parquet) by subject, using the metadata/.shards.json partition written by split_and_shard_subjects. For each (split, table) pair, every matching sub-shard is scanned, filtered to the split’s subject list (through the table’s subject_id expression), joined to any referenced table via _table.join, and written to data/<split>/<shard>/<table>.parquet. Rows are preserved unchanged — only re-organized — so this stage outputs raw source columns, not MEDS events.

This example uses the stage’s config.yaml file.

Input files:

data/patients/[0-2).parquet:
  MRN: [1, 2]
  dob: ["2000-01-01T00:00:00", "2001-02-02T00:00:00"]
  eye_color: ["BROWN", "BLUE"]

data/patients/[2-4).parquet:
  MRN: [3, 4]
  dob: ["2002-03-03T00:00:00", "2003-04-04T00:00:00"]
  eye_color: ["GREEN", "BROWN"]

data/labs/[0-2).parquet:
  patient_id: [1, 1]
  timestamp: ["2020-01-01T10:00:00", "2020-01-01T11:00:00"]
  test_name: ["HR", "TEMP"]
  result: [80.0, 36.6]

data/labs/[2-4).parquet:
  patient_id: [2, 2]
  timestamp: ["2020-01-02T12:00:00", "2020-01-02T13:00:00"]
  test_name: ["HR", "TEMP"]
  result: [75.0, 37.0]

data/labs/[4-6).parquet:
  patient_id: [3, 3]
  timestamp: ["2020-01-03T14:00:00", "2020-01-03T15:00:00"]
  test_name: ["HR", "TEMP"]
  result: [85.0, 36.5]

data/labs/[6-8).parquet:
  patient_id: [4, 4]
  timestamp: ["2020-01-04T09:00:00", "2020-01-04T10:00:00"]
  test_name: ["HR", "TEMP"]
  result: [70.0, 36.8]

metadata/.shards.json:
  train/0: [1, 4]
  tuning/0: [3]
  held_out/0: [2]

messy.yaml: |
  _defaults:
    subject_id: $MRN
  patients:
    eye_color:
      code: 'f"EYE_COLOR//{$eye_color}"'
      time: null
    dob:
      code: MEDS_BIRTH
      time: '$dob::"%Y-%m-%dT%H:%M:%S"'
  labs:
    _defaults:
      subject_id: $patient_id
    lab:
      code: $test_name
      time: '$timestamp::"%Y-%m-%dT%H:%M:%S"'
      numeric_value: $result

Expected output files:

data/train/0/patients.parquet:
  MRN: [1, 4]
  dob: ["2000-01-01T00:00:00", "2003-04-04T00:00:00"]
  eye_color: ["BROWN", "BROWN"]

data/train/0/labs.parquet:
  patient_id: [1, 1, 4, 4]
  timestamp:
    [
      "2020-01-01T10:00:00",
      "2020-01-01T11:00:00",
      "2020-01-04T09:00:00",
      "2020-01-04T10:00:00",
    ]
  test_name: ["HR", "TEMP", "HR", "TEMP"]
  result: [80.0, 36.6, 70.0, 36.8]

data/tuning/0/patients.parquet:
  MRN: [3]
  dob: ["2002-03-03T00:00:00"]
  eye_color: ["GREEN"]

data/tuning/0/labs.parquet:
  patient_id: [3, 3]
  timestamp: ["2020-01-03T14:00:00", "2020-01-03T15:00:00"]
  test_name: ["HR", "TEMP"]
  result: [85.0, 36.5]

data/held_out/0/patients.parquet:
  MRN: [2]
  dob: ["2001-02-02T00:00:00"]
  eye_color: ["BLUE"]

data/held_out/0/labs.parquet:
  patient_id: [2, 2]
  timestamp: ["2020-01-02T12:00:00", "2020-01-02T13:00:00"]
  test_name: ["HR", "TEMP"]
  result: [75.0, 37.0]

Run this stage:

MEDS_transform-stage <pipeline.yaml> convert_to_subject_sharded input_dir=<input> output_dir=<output>

aggregated_join

Demonstrates the aggregated form of _table.join (issue #65) — cols: {deathtime: min} groups the right-hand admissions table by the join key and reduces each listed column before the left join, so every patients row gains a single min(deathtime) value.

The right side is deliberately split across two chunk files: subject 111 has admissions in both chunks, and its earliest death-time (2020-03-01) lives in the second chunk — proving the aggregation runs over the concatenated multi-file scan, not per chunk. This is the MIMIC-IV fix_static_data replacement shape: earliest death-time per subject pulled from the admissions table straight in the MESSY spec.

This example uses the stage’s config.yaml file.

Input files:

data/patients/[0-2).parquet:
  subject_id: [111, 222]

data/admissions/[0-2).parquet:
  subject_id: [111, 111]
  deathtime: ["2020-03-05 00:00:00", "2020-03-03 00:00:00"]

data/admissions/[2-4).parquet:
  subject_id: [111, 222]
  deathtime: ["2020-03-01 00:00:00", "2021-06-01 00:00:00"]

metadata/.shards.json:
  train/0: [111]
  tuning/0: [222]

messy.yaml: |
  patients:
    _table:
      join:
        admissions:
          key: subject_id
          cols:
            deathtime: min
    death:
      code: MEDS_DEATH
      time: '$deathtime::"%Y-%m-%d %H:%M:%S"'

Expected output files:

data/train/0/patients.parquet:
  subject_id: [111]
  deathtime: ["2020-03-01 00:00:00"]

data/tuning/0/patients.parquet:
  subject_id: [222]
  deathtime: ["2021-06-01 00:00:00"]

Run this stage:

MEDS_transform-stage <pipeline.yaml> convert_to_subject_sharded input_dir=<input> output_dir=<output>