Data extraction (Databricks)¶

On the warehouse route, Registry Forge reads three CSV files exported from your clinical data warehouse. This page describes the Databricks code that writes them from typical SMART on FHIR ingestion tables, and how to adapt it to other warehouses.

PySpark notebook (databricks_export.py) SQL version (databricks_export.sql)

Run the export inside the approved environment

The three files hold every document and the link from document to patient, with names and birth dates if you include them. Write them only to storage your protocol and data use agreement approve, and move them to the environment where the pipeline runs by an approved route.

Output files¶

File	Columns	Each row
`ccda_chunks.csv`	`id`, `chunk_index`, `chunk_data`	Up to 30,000 characters of a Base64-encoded document
`fhir_chunks.csv`	`id`, `chunk_index`, `chunk_data`	Up to 30,000 characters of FHIR Bundle JSON
`uuid_mapping.csv`	`document_uuid`, `patient_id`, and optional `first_name`, `last_name`, `mrn`, `dob`, `gender`	One document or FHIR Bundle and the registry patient it belongs to

Put the two chunk files in <base-dir>/CCDA and FHIR data/ and uuid_mapping.csv in <base-dir>.

Why the content is split into rows¶

C-CDA documents and FHIR Bundles are often larger than 100 KB and can be several megabytes, while many warehouses limit the size of a CSV cell on export, often to about 32 KB. Splitting each document into rows keeps every cell under the limit. Chunk N always holds characters N × 30,000 to (N + 1) × 30,000 of the source, so sorting on (id, chunk_index) and joining the chunks gives back the document exactly.

The notebook¶

Edit the settings block at the top of databricks_export.py:

CCDA_TABLE        = "your_database.documents"
CCDA_ID_COL       = "document_uuid"
CCDA_CONTENT_COL  = "content_base64"
CCDA_FORMAT_COL   = "document_format"
CCDA_FORMAT_VALUE = "ccda"

FHIR_TABLE       = "your_database.fhir_bundles"
FHIR_ID_COL      = "bundle_uuid"
FHIR_CONTENT_COL = "bundle_json"
FHIR_PATIENT_COL = None        # set when the FHIR table names the registry patient

MAPPING_TABLE       = "your_database.documents"
MAPPING_DOC_ID_COL  = "document_uuid"
MAPPING_PATIENT_COL = "patient_id"
# optional demographic columns follow

OUTPUT_DIR = "/dbfs/FileStore/registry_forge_export"
CHUNK_SIZE = 30000

Set FHIR_PATIENT_COL when your FHIR table has a column naming the registry patient. The notebook then adds each FHIR Bundle to uuid_mapping.csv, and the pipeline assigns the Bundle to that patient whatever patient ID the health system wrote inside it. Without it, a Bundle's patient comes from the Patient resource inside the Bundle.

How the chunks are made¶

One Spark SQL expression writes one row per chunk without a Python user-defined function:

SELECT id, chunk_index, chunk_data
FROM (
  SELECT
    id,
    posexplode(
      transform(
        sequence(0, CAST(ceil(length(content) / 30000.0) AS INT) - 1),
        i -> substring(content, i * 30000 + 1, 30000)
      )
    ) AS (chunk_index, chunk_data)
  FROM source
)
ORDER BY id, chunk_index

ceil(length(content) / 30000) gives the number of chunks, sequence lists their indexes, transform cuts the slice for each index, and posexplode turns the list into one row per slice. The work stays inside Spark, so large tables split quickly.

CSV format¶

The notebook writes RFC 4180 CSV: every value is quoted, and a quote inside a value is written as two quotes (.option("escape", '"')). Spark's default writes a backslash before a quote instead, which pandas and the pipeline do not read, so keep that option if you change the code. The SQL version sets escape '"' in its OPTIONS for the same reason.

Spark writes each output as a folder holding a part file. The notebook's last cells move each part file to a plain ccda_chunks.csv, fhir_chunks.csv and uuid_mapping.csv, then read them back and print row counts and the number of distinct documents. The SQL version cannot rename files; rename the part files with the Databricks command-line tools or a shell.

Other warehouses¶

The pipeline does not depend on Databricks. Any engine that can produce id, chunk_index and chunk_data columns, sortable to rebuild each document, will do:

Warehouse	Functions to use
Snowflake	`LATERAL FLATTEN` over a generated array, or a recursive common table expression
BigQuery	`UNNEST(GENERATE_ARRAY(0, n))` with `SUBSTR()`
PostgreSQL	`generate_series(0, n)` with `substring()`
DuckDB	`range(0, n)` with `substr()`

Whatever the engine, write RFC 4180 CSV with doubled quotes.

Checking the export¶

Compare the counts the notebook prints with what you expect from the warehouse: the number of distinct id values in each chunk file should equal the number of documents and Bundles exported, and every document should have a row in uuid_mapping.csv. The ratio of chunk rows to distinct documents shows the typical document size; a high ratio means some very large documents, which work but take longer to reassemble.