chore: 🍱 split REDCap data into one Parquet file per form - #192
chore: 🍱 split REDCap data into one Parquet file per form#192martonvago wants to merge 22 commits into
Conversation
| common.json.write(field_metadata_preprocessed_path, field_metadata_preprocessed) | ||
|
|
||
|
|
||
| def task_split_forms( |
There was a problem hiding this comment.
I moved some complexity into this function as we wanted these tasks to do orchestration rather than just call another function that does the orchestration.
I do all reading and writing here to match previous tasks.
| form_to_fields = data.redcap.core.get_form_field_mapping( | ||
| common.json.read(field_metadata_path) | ||
| ) | ||
| form_to_events = data.redcap.core.get_form_event_mapping( | ||
| common.json.read(event_metadata_path) | ||
| ) | ||
| repeating_form_names = data.redcap.core.get_repeating_forms( | ||
| common.json.read(repeating_forms_path) | ||
| ) |
There was a problem hiding this comment.
This is the metadata we will use to split the data.
I could unite them into a single structure if people like that better, e.g.:
[
"bedq": {
"fields": [...],
"events": [...],
"repeats": False,
},
...
]
Could even be the output of the previous preprocessing step.
|
|
||
| @dataclass | ||
| class Form: | ||
| """Class to hold the name and data of a form.""" |
There was a problem hiding this comment.
I decided to do this instead of writing the form name into each df as a separate column only to drop that column later, as it felt a bit cleaner.
|
|
||
| def read_raw(raw_data_path: Path, form_to_fields: dict[str, list[str]]) -> pl.LazyFrame: | ||
| """Read the raw data into a LazyFrame with missing columns added.""" | ||
| raw_lf = pl.scan_csv(raw_data_path, infer_schema=False) |
There was a problem hiding this comment.
I'm reading into a lazy frame because the huge number of columns made the transformations very slow on data frames, even with only 2 raw batches. Lazy frames allow Polars to optimise operations a lot more.
| def read_raw(raw_data_path: Path, form_to_fields: dict[str, list[str]]) -> pl.LazyFrame: | ||
| """Read the raw data into a LazyFrame with missing columns added.""" | ||
| raw_lf = pl.scan_csv(raw_data_path, infer_schema=False) | ||
| return _with_missing_columns(raw_lf, form_to_fields) |
There was a problem hiding this comment.
I decided to add missing columns here, right at read-time. This means that we don't have to worry about this later on in the flow. Can be a separate step of course.
I also thought about extra columns (if a column is dropped later on in the study), but I don't think that's a problem. Any columns not in the latest metadata will not make it into staging, which feels like what we want.
Let me know if anyone has wise thoughts about these scenarios.
| return _with_missing_columns(raw_lf, form_to_fields) | ||
|
|
||
|
|
||
| def get_form_field_mapping( |
There was a problem hiding this comment.
These mapping functions use a for loop. I think this is the simplest and cleanest way of expressing the logic. But lmk if I should rewrite it.
|
|
||
| columns = [ | ||
| pl.col("record_id_s").alias("participant_id"), | ||
| pl.col("redcap_event_name").alias("event_id"), |
There was a problem hiding this comment.
This is where the other ids would be set up. (See PR description.)
| form_name, field_names = form_entry | ||
| events = form_to_events.get(form_name, []) | ||
| is_repeating = form_name in repeating_form_names | ||
| content_fields = so.keep(field_names, lambda field: field not in REDCAP_ID_COLS) |
There was a problem hiding this comment.
@signekb maybeee removing the admin fields could be part of this? We will include only the content_fields for each form, so we could strip out admin fields as well.
| ), | ||
| ] | ||
|
|
||
| return Form(name=form_name, data=raw_lf.filter(filters).select(columns).collect()) |
There was a problem hiding this comment.
This is where the data frame is materialised (i.e. the operations are executed).
lwjohnst86
changed the title
1 participant
Description
This PR splits REDCap data into one Parquet file per form. I am saying "form" intentionally because moving to resources will only come in the next step. Some forms correspond one-to-one to resources, but some are first transformed or joined (e.g. vas). For the first kind, we can just move the parquet files generated here straight over to staging. For the second kind, we will do the necessary transformations when staging.
Of interest is the handling of metadata fields/columns that are not present in some data batches. This happens when new columns are added, which are missing in old batches. To make all staging data have the same shape, I add the missing columns in this step (filled with null). This should work out nicely because, when we join staging batches, newer batches will trump older batches. So values in the new columns in newer batches will trump the null placeholders.
For now, this only deals with
participant_id,event_id, andsubmission_id.study_week_idand potentially other ids will fit roughly whereevent_idgoes. There's already plenty in this PR to discuss, so I think it's simpler to let thestudy_week_ididea mature and refactor later. Structurally, (participant_id,event_id,submission_id) works as a PK, so we can continue building the pipeline in the meantime.I tested this on GenomeDK and it ran in 50ish seconds, so that's rather slow. Room for optimisation...down to 5 with some magic and sorcery 🧙Closes #169
Needs a thorough review.
Checklist
just run-all