EHR Pipeline

Airflow DAG for processing OMOP EHR deliveries in Cloud Composer.

The DAG id is ehr-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 DAG 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, and related OMOP operations.
• OMOP_ANALYZER_ENDPOINT handles DQD, Achilles, Atlas table creation, and report generation.

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

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.

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
  dependencies/
    ehr/
      analysis.py
      bq.py
      constants.py
      dag_helpers.py
      file_config.py
      natural_keys.py
      omop.py
      participant_filter.py
      post_processing.py
      processing.py
      processing_jobs.py
      storage_backend.py
      utils.py
      validation.py
      vocab.py
      config/
        site_config.yml