chore: 🍱 split REDCap data into one Parquet file per form by martonvago · Pull Request #192 · onlimit-study/feasibility-data

martonvago · 2026-08-05T16:30:19Z

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, and submission_id. study_week_id and potentially other ids will fit roughly where event_id goes. There's already plenty in this PR to discuss, so I think it's simpler to let the study_week_id idea 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

Ran just run-all

…m name

martonvago · 2026-08-06T12:35:07Z

    common.json.write(field_metadata_preprocessed_path, field_metadata_preprocessed)


+def task_split_forms(


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.

martonvago · 2026-08-06T12:40:03Z

+    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)
+    )


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.

martonvago · 2026-08-06T12:49:26Z

+
+@dataclass
+class Form:
+    """Class to hold the name and data of a form."""


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.

martonvago · 2026-08-06T12:51:45Z

+
+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)


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.

This should be added as a comment, since it helps communicate why LazyFrame was used. Can you add that as a comment?

martonvago · 2026-08-06T12:55:09Z

+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)


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.

martonvago · 2026-08-06T13:01:09Z

+    return _with_missing_columns(raw_lf, form_to_fields)
+
+
+def get_form_field_mapping(


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.

It's hard to follow what's going on here and why these are needed, so it's hard for me to know what or if the loop is doing/how to improve on it. Could you refactor these to take a type/class and output a type/class that represents what you actually want/need from the forms? My instinct tells me this could be simpler but I can't pinpoint how yet.

Would this suggestion help?

martonvago · 2026-08-06T13:09:23Z

+
+    columns = [
+        pl.col("record_id_s").alias("participant_id"),
+        pl.col("redcap_event_name").alias("event_id"),


This is where the other ids would be set up. (See PR description.)

martonvago · 2026-08-06T13:12:45Z

+    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)


@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.

martonvago · 2026-08-06T13:15:47Z

+        ),
+    ]
+
+    return Form(name=form_name, data=raw_lf.filter(filters).select(columns).collect())


This is where the data frame is materialised (i.e. the operations are executed).

You might not have to collect at this stage, as writing to file will force it. At least that's how it could be done in R, and I know Polars has lots of similarities. But it might be different.

We have to collect before writing to check if the df is empty because we said we didn't want to create empty resources. Buuut we could actually allow empty resources in staging and let the properties extraction and batch joining mechanisms deal with them later. Or there are other alternatives to the is_empty check, such as writing without checking and deleting afterwards if the Parquet metadata says 0 rows.

In any case, we should definitely optimise by collecting all frames at once.

lwjohnst86

Nice start 🎉 Here's some initial comments and suggestions, haven't fully reviewed every bit.

lwjohnst86 · 2026-08-10T12:31:48Z

+
+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)


This should be added as a comment, since it helps communicate why LazyFrame was used. Can you add that as a comment?

lwjohnst86 · 2026-08-10T12:40:06Z

+    return _with_missing_columns(raw_lf, form_to_fields)
+
+
+def get_form_field_mapping(


It's hard to follow what's going on here and why these are needed, so it's hard for me to know what or if the loop is doing/how to improve on it. Could you refactor these to take a type/class and output a type/class that represents what you actually want/need from the forms? My instinct tells me this could be simpler but I can't pinpoint how yet.

lwjohnst86 · 2026-08-10T12:41:40Z

+    forms = so.fmap(
+        form_to_fields.items(),
+        lambda form_entry: _create_df_for_form(
+            form_entry, raw_lf, form_to_events, repeating_form_names
+        ),
+    )


I believe that Polars has a way to do this rather than use maps, which will probably save a lot of time. A quick look seems that maybe partition_by() might be it..? I know it's possible to do this in R, so I assume Polars can do it to.

I think partition_by is for grouping rows by the values in one or more columns, not grouping columns based on their names. I wasn't able to find a Polars-native way of doing what we want, but if you know what exactly you would use in R, maybe we could check whether that has an equivalent.

lwjohnst86 · 2026-08-10T12:43:53Z

+        ),
+    ]
+
+    return Form(name=form_name, data=raw_lf.filter(filters).select(columns).collect())


You might not have to collect at this stage, as writing to file will force it. At least that's how it could be done in R, and I know Polars has lots of similarities. But it might be different.

lwjohnst86 · 2026-08-10T12:47:46Z

+    for raw_data_path in raw_data_paths:
+        raw_data = data.redcap.core.read_raw(raw_data_path, form_to_fields)
+        forms = data.redcap.core.split_forms(
+            raw_data,
+            form_to_fields,
+            form_to_events,
+            repeating_form_names,
+        )
+
+        for form in forms:
+            data.redcap.core.write_form(form, forms_dir, raw_data_path)


This can be rewritten with map, which might allow us to use some parallel processing from the refactor, which will help speed things up.

There is a paralleliser plugin for Pytask, which could fit nicely here to parallelise the iterations of the outer loop (i.e. processing the raw batches). This would be equivalent to creating a map and executing in parallel.

Or do you mean something more specific by "the refactor"?

martonvago and others added 13 commits July 24, 2026 15:06

chore: 🏗️ add event and repeating forms metadata download to pytask

c865a74

chore: 🔧 update lockfile

e34ba7e

Merge branch 'main' into chore/pytask-other-metadata

23b4c6a

Merge branch 'main' into chore/pytask-other-metadata

ccff78c

Merge branch 'main' into chore/pytask-other-metadata

9e665a5

Merge branch 'main' into chore/pytask-other-metadata

3b0effb

Merge branch 'chore/pytask-other-metadata' into chore/split-forms

8928a40

chore: 🍱 split forms

cc620fa

refactor: ♻️ small tweaks

8c73cef

chore: 🔧 fix imports after merge

6482294

Merge branch 'chore/fix-imports' into chore/split-forms

e82797a

fix: 🐛 handle case when metadata has more columns than data

86e0332

Merge branch 'main' into chore/split-forms

730aed5

martonvago self-assigned this Aug 5, 2026

add-to-board-token Bot added this to Data development Aug 5, 2026

github-project-automation Bot moved this to Todo in Data development Aug 5, 2026

martonvago moved this from Todo to In Progress in Data development Aug 5, 2026

martonvago added 7 commits August 6, 2026 10:48

perf: ⚡️ use lazy frames

908bfd0

refactor: ♻️ use small dataclass instead of extra column to track for…

a4f28e5

…m name

refactor: ♻️ move metadata processing out of loop

14ee138

refactor: ♻️ move write call to build.py

8616731

refactor: ♻️ move read call to build.py

5fa8a88

refactor: ♻️ move column addition into read

8684da1

chore: 🔧 update lockfile

202e43f

martonvago commented Aug 6, 2026

View reviewed changes

refactor: ♻️ move timestamp extraction to core

6987d59

martonvago commented Aug 6, 2026

View reviewed changes

docs: 📝 tweak docstring

4bd0c4a

martonvago commented Aug 6, 2026

View reviewed changes

martonvago moved this from In Progress to In Review in Data development Aug 6, 2026

martonvago marked this pull request as ready for review August 6, 2026 13:22

martonvago requested a review from a team as a code owner August 6, 2026 13:22

lwjohnst86 changed the title ~~chore: 🍱 split REDCap data into one Parquet file per form~~ chore: 🍱 split REDCap data into one Parquet file per form Aug 6, 2026

K-Beicher mentioned this pull request Aug 7, 2026

Follow-up questions about source-dictionary.csv #183

Open

lwjohnst86 requested changes Aug 10, 2026

View reviewed changes

github-project-automation Bot moved this from In Review to In Progress in Data development Aug 10, 2026

		common.json.write(field_metadata_preprocessed_path, field_metadata_preprocessed)


		def task_split_forms(

		return _with_missing_columns(raw_lf, form_to_fields)


		def get_form_field_mapping(

Uh oh!

Conversation

martonvago commented Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Checklist

Uh oh!

martonvago Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

martonvago Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

martonvago Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

martonvago Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lwjohnst86 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

martonvago commented Aug 5, 2026 •

edited

Loading

martonvago Aug 6, 2026 •

edited

Loading

martonvago Aug 6, 2026 •

edited

Loading

martonvago Aug 6, 2026 •

edited

Loading

martonvago Aug 6, 2026 •

edited

Loading