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.yamlfile.
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.yamlfile.
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.yamlfile.
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: