EHR Pipeline

Airflow DAGs for processing OMOP EHR deliveries in Cloud Composer.

This repository defines two DAGs:

• ehr-pipeline (dags/ehr_pipeline.py) — processes each site's latest delivery independently. Covered by most of this document.
• omop-merge-pipeline (dags/omop_merge_pipeline.py) — combines every site's latest completed delivery into a single merged OMOP instance. See Merge Pipeline.

Purpose

For each site delivery, the pipeline:

1. Finds the most recent delivery folder in the site's GCS bucket.
2. Exports Connect reference data for the site.
3. Converts source files to Parquet.
4. Validates and normalizes OMOP files.
5. Upgrades delivered CDM versions when required.
6. Rewrites natural-key columns (PK + FK) so values are globally unique across sites.
7. Filters participants using Connect status data.
8. Runs vocabulary harmonization for supported clinical tables.
9. Generates derived tables.
10. Loads the processed data to BigQuery.
11. Runs DQD, Achilles, PASS, and delivery reporting.

Runtime Model

This repository contains the Composer DAGs and task helpers. The heavy processing is executed by external services:

• OMOP_PROCESSOR_ENDPOINT handles file conversion, validation, normalization, CDM upgrades, vocabulary work, BigQuery preparation, the merge extract/reconcile/build stages, and related OMOP operations.
• OMOP_ANALYZER_ENDPOINT handles DQD, Achilles, Atlas table creation, and report generation.

Each DAG coordinates site discovery, task ordering, retries, and pipeline logging.

Requirements

• Google Cloud Composer
• BigQuery datasets for CDM and analytics results
• Per-site GCS buckets with deliveries stored as top-level YYYY-MM-DD folders
• Reachable processor and analyzer services
• Composer worker permissions to:
  • read and write the configured GCS buckets
  • invoke the processor and analyzer services
  • create, truncate, and load BigQuery tables

Configuration

Site Config

The DAG reads site configuration from:

/home/airflow/gcs/dags/dependencies/ehr/config/site_config.yml

Example:

site:
  site_name:
    display_name: "Site Display Name"
    gcs_bucket: "my-site-bucket"
    file_delivery_format: ".csv"
    project_id: "gcp-project-id"
    cdm_bq_dataset: "omop_cdm"
    analytics_bq_dataset: "omop_analytics"
    cdm_version: "5.3"
    date_format: "%Y-%m-%d"
    datetime_format: "%Y-%m-%d %H:%M:%S"
    overwrite_site_vocab_with_standard: true
    site_connect_id: 123456789
    post_processing: 'remove_text-to-concept_measurements, some_other_task'

Fields used by the DAG:

• display_name: human-readable site name used in logs and reports

• gcs_bucket: bucket name only, without gs://

• file_delivery_format: expected source file format, typically .csv or .csv.gz

• project_id: GCP project for BigQuery operations

• cdm_bq_dataset: target OMOP dataset

• analytics_bq_dataset: target analytics dataset for DQD, Achilles, PASS, and Atlas outputs

• cdm_version: delivered CDM version

• date_format and datetime_format: source formatting hints for normalization

• overwrite_site_vocab_with_standard: when true, standard vocabulary is loaded and site vocab tables are skipped

• site_connect_id: Connect identifier used for site-level participant export

• post_processing: optional comma-delimited string of post-processing task names to run after vocabulary harmonization, written on a single line. Each name must correspond to a SQL file at reference/sql/post_processing/<task_name>.sql in the OMOP file processor repo. Tasks for a given site run in series in the order they appear in the string; different sites' tasks run in parallel.

  post_processing: 'remove_text-to-concept_measurements, some_other_task'

  Omit the field or set it to an empty string to skip post-processing for the site.

Environment Variables

Defined in dags/dependencies/ehr/constants.py:

• OMOP_PROCESSOR_ENDPOINT
• OMOP_ANALYZER_ENDPOINT
• OMOP_TARGET_VOCAB_VERSION
• OMOP_TARGET_CDM_VERSION
• OMOP_VOCAB_GCS_PATH
• CONNECT_DATASET_ID

Read in dags/dependencies/ehr/storage_backend.py:

• STORAGE_BACKEND — gcs (default) or local. The local backend is intended for development against locally mounted files; production should leave this unset.

Defined in dags/dependencies/ehr/merge/constants.py (merge pipeline only; each has a default):

• MERGE_BUCKET — GCS bucket for the merged instance (default ehr_merged)
• MERGE_CDM_DATASET — target BigQuery CDM dataset for the merged instance (default ehr_merged)
• MERGE_ANALYTICS_DATASET — target analytics dataset for the merged instance's DQD, Achilles, PASS, and Atlas outputs (default ehr_merged)
• MERGE_SITE_NAME — synthetic site_name the merge run logs under in the shared BigQuery logging table; also the @SITE salt for the merged instance's surrogate keys (default ehr_merged)
• GCP_PROJECT_ID — project hosting the merged datasets and the merge Cloud Run jobs (default nih-nci-dceg-connect-dev)

DAG Flow

The main flow in dags/ehr_pipeline.py is:

1. check_api_health
2. id_sites_to_process
3. get_unprocessed_files
4. process_delivery task group, per file in order: convert_file → validate_file → normalize_file → cdm_upgrade → per-site populate_cdm_source_file → per-file globalize_natural_keys
5. filtering task group: per-site retrieve_connect_data → per-file filter_participants
6. vocab_harmonization task group
7. post_processing task group
8. generate_derived_tables task group
9. load_to_bigquery task group
10. cleanup
11. reporting task group: per-(site, artifact_type) generate_report_artifact → per-site consolidate_report
12. analysis task group: per-site dqd, achilles, and pass_analysis run in parallel; once all three finish, per-site atlas_results_tables and generate_delivery_report run
13. mark_delivery_complete
14. log_done

Vocabulary Harmonization

The vocab_harmonization task group runs these steps in order:

1. harmonize_vocab_source_target
2. harmonize_vocab_target_remap
3. harmonize_vocab_source_concept_backfill
4. harmonize_vocab_domain_check
5. harmonize_vocab_omop_etl
6. harmonize_vocab_consolidate
7. harmonize_vocab_discover_tables
8. flatten_table_configs
9. harmonize_vocab_deduplicate_table

Only tables listed in VOCAB_HARMONIZED_TABLES are processed by these steps.

Two additional file-processor harmonization steps — harmonize_vocab_target_replacement and harmonize_vocab_secondary_concept_backfill — are defined as Airflow tasks but are intentionally excluded from the DAG's >> chain (see the comment near the vocab_harmonization task group). They remain callable on the file processor as endpoints but are not invoked by the current DAG.

Post Processing

The post_processing task group sits between vocabulary harmonization and derived-table generation, so derived tables reflect the final post-processed state of the harmonized clinical tables.

Per-site serial, cross-site parallel. The task group maps one Airflow instance per site that has at least one post-processing task configured. Inside each instance, the configured tasks run sequentially in the order they appear in site_config.yml — the file processor's /post_processing endpoint is invoked once per task, and the next task does not start until the previous one returns. This matters because post-processing tasks share the same parquet artifacts; later tasks may depend on the state produced by earlier ones. Different sites' chains run independently in parallel.

If any task in a site's chain fails, the remaining tasks for that site are not attempted. Empty or whitespace-only entries in the comma-delimited string are filtered out by SiteConfig.post_processing, so stubs like post_processing: '' collapse to no work. When no site has any tasks configured, the mapped task expands to zero instances and the DAG flows straight through.

Each task name in the site config must map to a SQL file at reference/sql/post_processing/<task_name>.sql in the OMOP file processor repo. The processor's /post_processing endpoint returns 400 if the script is missing or if the script attempts to write to a vocabulary file; the DAG task fails fast in either case.

Derived Tables

Derived tables are generated after post-processing and before BigQuery loading.

The current derived tables are defined by DERIVED_DATA_TABLES in dags/dependencies/ehr/constants.py.

Merge Pipeline

DAG id omop-merge-pipeline (dags/omop_merge_pipeline.py). It combines every site's latest completed delivery into a single merged OMOP instance, then loads and analyzes that instance the same way the single-site pipeline does. The DAG is manual/triggered only (schedule_interval=None).

This is v1 of the merge functionality. It does not "look back" to recover data for participants who have withdrawn but not requested data deletion; those participants are removed by re-applying the existing Connect participant filter against a fresh all-sites participant status.

Merged Instance

The merged instance lives at a normal {bucket}/{delivery_date} address so every processor path builder, the BigQuery log, and the participant filter work unchanged. Its identity is resolved once by resolve_merge_instance and shared via XCom (MergeInstance):

• bucket = MERGE_BUCKET
• delivery_date = the DAG run's logical date as a US/Eastern calendar date, so a run triggered late-evening US time does not land in the next day's folder
• CDM / analytics datasets = MERGE_CDM_DATASET / MERGE_ANALYTICS_DATASET
• logged site_name = MERGE_SITE_NAME

The merge reuses the same site_config.yml as ehr-pipeline to enumerate sites and resolve each site's source bucket and display name.

Group Selection and Validation

select_merge_groups builds one group per site — each site's latest completed delivery, discovered via the processor's /get_latest_completed_delivery. Sites with no completed delivery are logged and skipped. (Group selection is parameterized for pins and explicit site lists to support a future v2, but v1 always takes every site's latest completed delivery.)

validate_mergeable then reads each selected delivery's standardized CDM version from its processed cdm_source (/get_delivery_cdm_version) and requires every group to match OMOP_TARGET_CDM_VERSION:

• Any CDM-version conflict is a hard fail — there is no auto-upgrade in the merge.
• Only the "nothing to merge" case (no groups) short-circuits the DAG cleanly.
• Vocabulary version is not checked; merged tables are re-harmonized to the target vocabulary.

Merged Tables

MERGED_TABLES (in dags/dependencies/ehr/merge/constants.py) is the set extracted and reconciled from each delivery:

• person plus the vocab-harmonized clinical tables are sourced from each delivery's already-harmonized artifacts/omop_etl/ output and re-harmonized on the merged instance.
• Non-harmonized supporting tables (e.g. location, provider, visit_detail, death, cost) are plain-unioned from artifacts/converted_files/ as-is.

A (site, table) whose source parquet does not exist emits no chunk, so the extract job is never asked for a missing file. care_site and cdm_source are built de novo (not carried over from deliveries), the standard target vocabulary is loaded fresh, and the derived era / observation-period tables are regenerated after harmonization.

DAG Flow

The main flow in dags/omop_merge_pipeline.py is:

1. check_api_health
2. build_optimized_vocab, resolve_merge_instance, select_merge_groups (in parallel)
3. validate_mergeable (short-circuit gate)
4. provision_merge_instance — creates the merged artifact directories (including merge_chunks/) and logs the run start
5. extract task group: build_extract_configs (groups × MERGED_TABLES, existence-filtered) → extract_participant_chunk mapped per (group, table), writing provenance-named chunks to merge_chunks/<table>/. For person, extract stamps care_site_id with a hash of the source site's display name.
6. reconcile task group: build_reconcile_configs → reconcile_table mapped per table, unioning each table's chunks into converted_files/<table>.parquet
7. retrieve_merged_connect_data — exports one all-sites participant_status.parquet to the merged instance
8. filtering task group: filter_participants mapped per reconciled table, re-applying the Connect filter in place against the fresh all-sites status
9. vocab_harmonization task group — the same live harmonization chain as ehr-pipeline, retargeted at the merged instance (non-clinical tables auto-skip)
10. build_merge_care_site and build_merge_cdm_source — de novo care_site and cdm_source, built after reconcile
11. generate_derived_tables task group — era / observation-period tables from the merged harmonized data
12. load_to_bigquery task group: prepare_merge_bq → load_merge_harmonized_tables → load_merge_target_vocab → load_merge_table (non-harmonized tables) → load_merge_care_site → load_merge_cdm_source → load_merge_derived_tables → create_merge_missing_tables
13. analysis task group (merge_dqd, merge_achilles, merge_pass in parallel, then merge_create_atlas_tables) and generate_merge_report run in parallel
14. mark_merge_complete
15. log_done — DAG-level failure backstop (ALL_DONE); fails the run and logs an error row if any task failed

The extract and reconcile stages run as Cloud Run jobs on the file processor (ccc-omop-file-processor-extract-participant-chunk-job, ccc-omop-file-processor-reconcile-chunks-job), since the merged files are too large for the service's 1-hour timeout.

Deployment

1. Upload dags/ to the Composer DAGs bucket, preserving paths.
2. Ensure site_config.yml is present at the configured path in the DAGs bucket.
3. Set the required environment variables in Composer.
4. Grant Composer access to the configured GCS buckets, BigQuery datasets, and service endpoints.
5. Trigger ehr-pipeline from Airflow.

Operational Notes

• A site is processed when its latest delivery has no log row or its most recent log row is in an error state.
• load_harmonized_tables skips cleanly when no harmonized clinical tables were produced.
• load_remaining skips:
  • vocabulary tables, but only when overwrite_site_vocab_with_standard is true (i.e., when the DAG is loading the standard target vocabulary instead of site-delivered vocab)
  • tables listed in VOCAB_HARMONIZED_TABLES, since those are loaded earlier from artifacts/omop_etl/
  • cdm_source, which is loaded during cleanup
• log_done inspects task states and fails the DAG if any upstream task failed, even if a mapped task failure was otherwise easy to miss.

Repository Layout

dags/
  ehr_pipeline.py
  omop_merge_pipeline.py
  dependencies/
    ehr/
      analysis.py
      bq.py
      constants.py
      dag_helpers.py
      file_config.py
      processing_jobs.py
      processor_api.py
      storage_backend.py
      utils.py
      vocab.py
      merge/
        builders.py
        constants.py
        groups.py
        identity.py
        processing_jobs.py
        validation.py
      config/
        site_config.yml