diff --git a/dags/non_program/dewrangle_id_minting/global_id_minting.py b/dags/non_program/dewrangle_id_minting/global_id_minting.py new file mode 100644 index 0000000..0bc7cb7 --- /dev/null +++ b/dags/non_program/dewrangle_id_minting/global_id_minting.py @@ -0,0 +1,169 @@ +"""DAG to mint Global IDs in Dewrangle for a given PostgreSQL table with descriptors.""" + +from datetime import datetime +import csv +import logging + +from airflow.models.dag import DAG +from airflow.models import Param +from airflow.operators.python import PythonOperator +from airflow.operators.bash import BashOperator + +try: + from airflow.sdk.bases.hook import BaseHook +except ( + ImportError +): # Since Airflow 3.1, the BaseHook is in the airflow.sdk.bases.hook module + from airflow.hooks.base import BaseHook +import psycopg2 + +logger = logging.getLogger(__name__) + + +with DAG( + dag_id="global_id_minting", + description=""" + Given a PostgreSQL table with descriptors, generate global IDs and load + those IDs into a target table. + """, + start_date=datetime(2026, 6, 1), + schedule=None, + catchup=False, + params={ + "descriptor_schema_name": Param( + default="default_schema", + type="string", + title="Descriptor Schema Name", + description="Schema name where the table with descriptors that need global IDs is located", + ), + "descriptor_table_name": Param( + default="default_table", + type="string", + title="Descriptor Table Name", + description="Name of the table with descriptors that need global IDs", + ), + "globalid_schema_name": Param( + default="default_schema", + type="string", + title="Global ID Schema Name", + description="Schema name where the table with generated global IDs is located", + ), + "globalid_table_name": Param( + default="default_table", + type="string", + title="Global ID Table Name", + description="Name of the table with generated global IDs", + ), + "create_new_globalid_table": Param( + default=False, + type="boolean", + title="Create New Global ID Table", + description="Whether to create a new table for the generated global IDs or not", + ), + "env": Param( + default="qa", + type="string", + title="Environment", + description="Environment for the global ID minting command", + enum=["qa", "prod"], + ), + "dewrangle_organization_id": Param( + default="T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", + type="string", + title="Dewrangle Organization ID", + description="Organization ID for the dewrangle global ID minting command", + enum=[ + "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=", + "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=", + "T3JnYW5pemF0aW9uOmNsZWhibTF4ZjAwZTdpY2VzZjI0d2tlNHk=", + ], + values_display={ + "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=": "dff dev", + "T3JnYW5pemF0aW9uOmNsZHN4MzRrbjAwMTRnMGVzY3JndzUzYWQ=": "Kids First DRC", + "T3JnYW5pemF0aW9uOmNsZWhibTF4ZjAwZTdpY2VzZjI0d2tlNHk=": "INCLUDE DCC", + }, + ), + }, + tags=["non_program", "id_minting", "dewrangle"], +) as dag: + + # Get Airflow connection + postgres_conn = BaseHook.get_connection("postgres_prd_svc") + dewrangle_conn = BaseHook.get_connection("dewrangle_api") + + def read_table_to_file(conn=postgres_conn, **context): + """Read from PostgreSQL table and save to local file.""" + schema_name = context["params"]["descriptor_schema_name"] + table_name = context["params"]["descriptor_table_name"] + + # Connect to PostgreSQL + connection = psycopg2.connect( + host=conn.host, + port=conn.port or 5432, + database=conn.schema, + user=conn.login, + password=conn.password, + ) + + cursor = connection.cursor() + + # Query the table + query = f"SELECT * FROM {schema_name}.{table_name}" + cursor.execute(query) + + # Get column names + column_names = [desc[0] for desc in cursor.description] + rows = cursor.fetchall() + + logger.info(f"Column names: {column_names}") + logger.info(f"Number of rows: {len(rows)}") + + # Save to local file + output_file = f"/tmp/{schema_name}_{table_name}_export.csv" + with open(output_file, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(column_names) + writer.writerows(rows) + + cursor.close() + connection.close() + + logger.info(f"Data exported to {output_file}") + return output_file + + read_and_export = PythonOperator( + task_id="read_and_export", + python_callable=read_table_to_file, + op_kwargs={ + "conn": postgres_conn, + }, + ) + + {set create_dewrangle_ids_table_flag = "--create-dewrangle-ids-table" if dag.params["create_new_globalid_table"] else ""} + + create_dewrangle_ids_table_flag = ( + "--create-dewrangle-ids-table" + if dag.params["create_new_globalid_table"] + else "" + ) + + mint_ids = BashOperator( + task_id="mint_ids", + bash_command="${ID_MINTING_PATH}/bin/d3b-dewrangle global-id-mint --env {{ params.env }} --db dcc --organization-id {{ params.dewrangle_organization_id }} --manifest {{ ti.xcom_pull(task_ids='read_and_export') }} {{ create_dewrangle_ids_table_flag }}", + env={ + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA": "{{ params.globalid_schema_name }}", + "QA_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE": "{{ params.globalid_table_name }}", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_SCHEMA": "{{ params.globalid_schema_name }}", + "PROD_DCC_WAREHOUSE_DEWRANGLE_IDS_TABLE": "{{ params.globalid_table_name }}", + "DCC_WAREHOUSE_HOST": postgres_conn.host, + "DCC_WAREHOUSE_PORT": str(postgres_conn.port or 5432), + "DCC_WAREHOUSE_DB_NAME": postgres_conn.schema, + "DCC_WAREHOUSE_DB_USER": postgres_conn.login, + "DCC_WAREHOUSE_DB_USER_PW": postgres_conn.password, + "DEWRANGLE_BASE_URL": dewrangle_conn.host, + "DEWRANGLE_TOKEN": dewrangle_conn.password, + "DEWRANGLE_CLIENT_EXECUTION_TIMEOUT": "300", + }, + ) + + read_and_export >> mint_ids diff --git a/pipeline_docs/guides/how_to_mint_global_ids.md b/pipeline_docs/guides/how_to_mint_global_ids.md new file mode 100644 index 0000000..1d5180d --- /dev/null +++ b/pipeline_docs/guides/how_to_mint_global_ids.md @@ -0,0 +1,207 @@ +# How to Mint Global Identifiers in Airflow + +This guide explains how to mint global identifiers with the Airflow DAG +`global_id_minting`. + +The DAG performs two steps: + +1. exports all rows from a source warehouse table to a temporary CSV file +2. runs `d3b-dewrangle global-id-mint` against that CSV and loads the minted + IDs into a target warehouse table + +Global ID's are generally minted at some point in harmonization of a study at +the discretion of the harmonizer. + +## Before you start + +Make sure all of the following are true before triggering the DAG: + +1. You can access the hosted Airflow instance. If not, complete the setup in + [connect-to-airflow.md](/Users/friedmanc1/Documents/include-dbt-sandbox/pipeline_docs/guides/connect-to-airflow.md). +2. The DAG `global_id_minting` is available in Airflow. +3. Your source table already exists in the warehouse. +4. You know which warehouse schema and table should receive the minted global + IDs. Note that this table does not need to exist. +5. Your source table columns match the descriptor format expected by + `d3b-dewrangle global-id-mint`. Specifically, the columns required are + `fhirResourceType`, `descriptor`, and `descriptorState`. To mint IDs within + a specific study, the column `studyGlobalId` is also required. + +## What the DAG does + +When you trigger the DAG, Airflow: + +1. Reads every column and row from + `{descriptor_schema_name}.{descriptor_table_name}`. +2. Writes the result to a temporary file in `/tmp` on the worker. +3. Runs: + +```bash +d3b-dewrangle global-id-mint \ + --env \ + --db dcc \ + --organization_id \ + --manifest /tmp/__export.csv +``` + +1. Loads the minted identifiers into the target table named by + `globalid_schema_name` and `globalid_table_name`. + +## Parameters to provide + +Trigger the DAG with the following parameters below. While there is the option +to enter each parameter individually, airflow allows entering these parameters +as a JSON object: + +```json +{ + "env": "qa", + "globalid_table_name": "global_ids", + "globalid_schema_name": "friedmanc1_dev_schema", + "descriptor_table_name": "dewrangle_id_minting_test", + "descriptor_schema_name": "friedmanc1_dev_schema", + "dewrangle_organization_id": "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=" +} +``` + +### `descriptor_schema_name` + +The schema that contains the descriptor table to export. + +Example: + +```json +"descriptor_schema_name": "public" +``` + +### `descriptor_table_name` + +The table that contains the descriptors that need global identifiers. + +Example: + +```json +"descriptor_table_name": "study_subject_descriptors" +``` + +### `globalid_schema_name` + +The schema where the minted identifier table should be written. + +Example: + +```json +"globalid_schema_name": "access" +``` + +### `globalid_table_name` + +The destination table name for minted identifiers. + +Example: + +```json +"globalid_table_name": "subject_global_ids" +``` + +### `env` + +The runtime environment passed to `d3b-dewrangle`. + +Allowed values: + +1. `qa` +2. `prod` + +Example: + +```json +"env": "qa" +``` + +### `dewrangle_organization_id` + +The dewrangle organization identifier used for minting. + +The DAG currently exposes these labeled options in Airflow: + +1. `dff dev` +2. `Kids First DRC` +3. `INCLUDE DCC` + +Choose the organization that should own the minted identifiers. Note that these +options are human-readable pointers to the organization ID that dewrangle uses. + +## Triggering the DAG + +1. Open the hosted Airflow UI. +2. Search for the DAG `global_id_minting`. +3. Open the DAG details page. +4. Click `Trigger DAG`. +5. Replace the default parameters with your run configuration. +6. Start the run. + +Example parameter payload: + +```json +{ + "descriptor_schema_name": "public", + "descriptor_table_name": "study_subject_descriptors", + "globalid_schema_name": "access", + "globalid_table_name": "subject_global_ids", + "env": "qa", + "dewrangle_organization_id": "T3JnYW5pemF0aW9uOmNtZjJ3bzlrdDAwMG9rMTAxcHd4cHFmMWQ=" +} +``` + +## How to monitor the run + +The DAG has two tasks: + +1. `read_and_export` +2. `mint_ids` + +Use task logs to diagnose failures. + +### If `read_and_export` fails + +Check for: + +1. an incorrect `descriptor_schema_name` +2. an incorrect `descriptor_table_name` +3. missing warehouse permissions +4. descriptor data that cannot be exported cleanly + +### If `mint_ids` fails + +Check for: + +1. an invalid `dewrangle_organization_id` +2. a mismatch between your descriptor CSV columns and what + `d3b-dewrangle global-id-mint` expects +3. a target schema or table name that should not be used for the selected run +4. downstream dewrangle or warehouse connectivity issues + +## Verifying results + +After the DAG succeeds: + +1. query the target table `{globalid_schema_name}.{globalid_table_name}` +2. confirm the expected number of rows were written +3. validate that the minted global identifiers match the descriptors you + supplied + +## Operational notes + +The current DAG implementation has a few important behaviors to keep in mind: + +1. The source export uses `SELECT *`, so the full source table is exported. +2. The exported CSV is written to a temporary file on the Airflow worker. +3. The DAG reads warehouse credentials from the Airflow connection + `postgres_dev_svc`. +4. The `env` parameter changes the dewrangle CLI flag, but the DAG code still + sources its warehouse connection details from the same Airflow connection. + +If you need production minting behavior that differs from the current DAG +implementation, review the DAG configuration before running it with `env` set +to `prod`.