337 lines
13 KiB
Markdown
337 lines
13 KiB
Markdown
# MeDaX Pipeline
|
||
|
||
## 📋 Description
|
||
The MeDaX pipeline transforms healthcare data from FHIR databases into Neo4j graph databases. This conversion enables efficient searching, querying, and analyses of interconnected health data that would otherwise be complex to retrieve using traditional SQL databases.
|
||
|
||
## ✨ Features
|
||
- Seamless conversion from FHIR to Neo4j graph structure
|
||
- Support for patient-centric data retrieval using FHIR's `$everything` operation
|
||
- Configurable batch processing for handling large datasets
|
||
- Docker-based deployment for easy setup and portability
|
||
- Compatible with public FHIR servers (e.g., HAPI FHIR) and private authenticated instances
|
||
- Optional post-processing queries for patient timelines and shared clinical concepts
|
||
|
||
## ⚙️ Prerequisites
|
||
- [Docker](https://docs.docker.com/engine/install/) with the [Docker Compose plugin](https://docs.docker.com/compose/install/linux/)
|
||
- A FHIR database with API access and the `$everything` operation enabled for retrieving patient data
|
||
- Alternatively: Use a public FHIR server such as [HAPI FHIR](https://hapi.fhir.org/) (default configuration)
|
||
- Optional: [Semspect](https://www.semspect.de/) license
|
||
- free scientific licenses can be requested
|
||
- helps with graph exploration
|
||
|
||
## 🚀 Installation
|
||
|
||
### Setup
|
||
1. Clone this repository
|
||
2. Create an environment configuration file
|
||
3. Configure the environment variables in `.env`:
|
||
- For HAPI test server (default): No changes needed
|
||
- For custom FHIR server:
|
||
- Change `MODE` to anything else
|
||
- Uncomment and set `URL`, `PASSWORD`, and `USERNAME` variables
|
||
- Adjust `BATCH_SIZE` and `NUMBER_OF_PATIENTS` according to your needs
|
||
- Configure any required proxy settings
|
||
|
||
4. If needed, modify proxy settings in the `Dockerfile`
|
||
- Uncomment and set proxy variables
|
||
|
||
### Running the Pipeline
|
||
|
||
#### Option A: Connect to an existing FHIR server (e.g. HAPI)
|
||
**Start the containers:**
|
||
```bash
|
||
docker compose up --build
|
||
```
|
||
|
||
**Stop and clean up (between runs):**
|
||
```bash
|
||
docker compose down --volumes
|
||
```
|
||
|
||
**Complete removal (containers and images):**
|
||
```bash
|
||
docker compose down --volumes --rmi all
|
||
```
|
||
|
||
#### Option B: Build a local FHIR server with POLAR data
|
||
This spins up a local [Blaze](https://github.com/samply/blaze) FHIR server pre-loaded with synthetic POLAR test bundles, using the `blaze` Compose profile.
|
||
|
||
**Setup:** in your `.env`, set:
|
||
```
|
||
COMPOSE_PROFILES=blaze
|
||
FHIR_SERVER_URL=http://blaze:8080/fhir
|
||
```
|
||
|
||
**Start the containers:**
|
||
```bash
|
||
docker compose up --build
|
||
```
|
||
|
||
**Stop and clean up (between runs):**
|
||
```bash
|
||
docker compose down --volumes
|
||
```
|
||
|
||
**Complete removal (containers and images):**
|
||
```bash
|
||
docker compose down --volumes --rmi all
|
||
```
|
||
|
||
#### Option C: Build a local FHIR server with POLAR data and a local keycloak server
|
||
This builds on Option B and adds a local Keycloak instance plus an OAuth2 proxy in front of Neo4j, using the `blaze`, `keycloak`, and `proxy` Compose profiles.
|
||
|
||
**Setup:** in your `.env`, set:
|
||
```
|
||
COMPOSE_PROFILES=blaze,keycloak,proxy
|
||
FHIR_SERVER_URL=http://blaze:8080/fhir
|
||
```
|
||
|
||
**Start the containers:**
|
||
##### Init containers
|
||
```bash
|
||
docker compose up --build
|
||
```
|
||
##### Create realm, client and user
|
||
1. Open keycloak in a browser: localhost:4040 and login with user: admin and password: admin
|
||
2. Create the realm "testPipeline"
|
||
3. Create a client:
|
||
* Client ID: neo4j-client
|
||
* Always display in UI "on"
|
||
* Hit the "Next" button
|
||
* Client authentification: on
|
||
* Hit the "Next" button
|
||
* Root URL: http://localhost:8082
|
||
* Home URL: http://localhost:8082
|
||
* Valid redirects URIs: http://localhost:8082/oauth2/callback
|
||
* Valid post logout redirect URIs: http://localhost:8082/*
|
||
* Web origin: http://localhost:8082
|
||
* Hit the "Save" button
|
||
* Go to the "Credentials" tab
|
||
* Copy the "Client Secret" to "NEO4J_OAUTH_SECRET" in your .env
|
||
4. Create a user:
|
||
* "Email verified": on
|
||
* Choose a user name
|
||
* Use random mail, e.g.: test@example.com
|
||
* Hit the "Create button"
|
||
* Click on the new user, go to the "Credentials" tab
|
||
* Set a password and disable "Temporary"
|
||
|
||
##### Restart containers without destroying the volumes
|
||
```bash
|
||
docker compose down
|
||
docker compose up
|
||
```
|
||
|
||
**Stop**
|
||
```bash
|
||
docker compose down
|
||
```
|
||
> **Note:** When adding --volumes to down, the keycloak setup will be lost
|
||
|
||
**Complete removal (containers and images):**
|
||
```bash
|
||
docker compose down --volumes --rmi all
|
||
```
|
||
|
||
> **Note:** Depending on your Docker installation, you might need to use `docker-compose` instead of `docker compose`.
|
||
> **Note:** With `COMPOSE_PROFILES` set in your `.env`, every `docker compose` command automatically includes the right services - no `-f` files needed.
|
||
|
||
## 🔍 Accessing the Neo4j Database
|
||
|
||
Once the pipeline has completed processing, you can access the Neo4j database:
|
||
|
||
1. Open your browser and navigate to `http://localhost:8080/` or `http://localhost:8082/` if you used keycloak (Option C, served via the `proxy` profile's `neo4j-proxy` service)
|
||
2. Connect by clicking the button, user name and password are not needed (disabled by config)
|
||
|
||
## 📊 Example Queries
|
||
|
||
Here are some basic Cypher queries to get you started with exploring your health data:
|
||
|
||
```cypher
|
||
// Count all nodes by type
|
||
MATCH (n) RETURN labels(n) as NodeType, count(*) as Count;
|
||
|
||
// Find all records for a specific patient
|
||
MATCH (p:Patient {id: 'patient-id'})-[r]-(connected)
|
||
RETURN p, r, connected;
|
||
|
||
// Retrieve all medication prescriptions
|
||
MATCH (m:Medication)-[r]-(p:Patient)
|
||
RETURN m, r, p;
|
||
```
|
||
|
||
## 🛠️ Post-processing Options
|
||
|
||
After the import has finished, you can optionally enrich the graph with additional structure. The queries below are run manually against the Neo4j database, e.g. by pasting them into the Neo4j Browser (see [Accessing the Neo4j Database](#-accessing-the-neo4j-database)).
|
||
|
||
| Step | Adds | Purpose |
|
||
|------|------|---------|
|
||
| 1. Patient timeline | `SAME_TIME`, `NEXT_EVENT` relationships | Orders each patient's clinical events chronologically |
|
||
| 2. Concept nodes | `:Concept` nodes, `MAPS_TO` relationships | Merges identical codings into shared concept nodes |
|
||
| 3. Patient–concept links | `HAS_CONCEPT` relationships | Connects patients directly to the concepts in their record |
|
||
|
||
**Notes:**
|
||
- Step 2 requires the [APOC](https://neo4j.com/docs/apoc/current/) plugin (`apoc.text.join`, `apoc.coll.sort`).
|
||
- Step 3 depends on Step 2 and must be run afterwards. Step 1 is independent.
|
||
- All queries use `MERGE`, so re-running them does not create duplicate nodes or relationships.
|
||
- On large databases these queries can take a while; consider running them on a subset of patients first.
|
||
|
||
### 1. Patient Timeline Edges
|
||
|
||
Builds a per-patient timeline from `Condition`, `Observation` and `Procedure` events, using their respective timestamps (`recordedDate`, `effectiveDateTime`, `performedDateTime`). Events without a timestamp are skipped.
|
||
|
||
- Events sharing the same timestamp are grouped; the first event of each group is linked to the others via `SAME_TIME`.
|
||
- The first events of consecutive groups are linked via `NEXT_EVENT`, forming a chronological chain per patient.
|
||
|
||
```cypher
|
||
// CREATE PATIENT TIMELINE EDGES
|
||
|
||
// Iterate over all patients
|
||
MATCH (p:Patient)
|
||
|
||
// Get events for each patient
|
||
OPTIONAL MATCH (p)<--(e)
|
||
WHERE e:Condition OR e:Observation OR e:Procedure
|
||
|
||
// Extract the relevant datetime property for each event
|
||
WITH p, e,
|
||
CASE
|
||
WHEN e:Condition THEN e.recordedDate
|
||
WHEN e:Observation THEN e.effectiveDateTime
|
||
WHEN e:Procedure THEN e.performedDateTime
|
||
END AS eventTime
|
||
WHERE eventTime IS NOT NULL
|
||
|
||
// Normalize time format
|
||
WITH p, e, datetime(eventTime) AS dt
|
||
ORDER BY p, dt ASC
|
||
|
||
// Group events per patient and per timestamp
|
||
WITH p, dt, collect(e) AS events
|
||
ORDER BY p, dt ASC
|
||
WITH p, collect({time: dt, events: events}) AS timeline
|
||
|
||
// Create SAME_TIME edges
|
||
CALL {
|
||
WITH timeline
|
||
UNWIND timeline AS group
|
||
WITH group.events AS events
|
||
WITH events[0] AS firstNode, events[1..] AS others
|
||
UNWIND others AS o
|
||
MERGE (firstNode)-[:SAME_TIME]->(o)
|
||
}
|
||
|
||
// Create NEXT_EVENT
|
||
CALL {
|
||
WITH timeline
|
||
UNWIND range(0, size(timeline)-2) AS i
|
||
WITH timeline[i].events[0] AS firstCurr,
|
||
timeline[i+1].events[0] AS firstNext
|
||
MERGE (firstCurr)-[:NEXT_EVENT]->(firstNext)
|
||
}
|
||
```
|
||
|
||
**Example:** walk a patient's timeline in order:
|
||
```cypher
|
||
MATCH path = (start)-[:NEXT_EVENT*]->(end)
|
||
WHERE (start)-->(:Patient {id: 'patient-id'})
|
||
AND NOT ()-[:NEXT_EVENT]->(start)
|
||
AND NOT (end)-[:NEXT_EVENT]->()
|
||
RETURN path;
|
||
```
|
||
|
||
### 2. Concept Nodes
|
||
|
||
Collects all codings (`*_code`, `*_system`, `*_display` property triples) of each `Code` node and creates a shared `Concept` node for every distinct combination of `system::code` pairs. Resources coded identically — across patients — then point to the same concept.
|
||
|
||
- `Concept.id`: sorted `system::code` pairs joined with `||`
|
||
- `Concept.displays`: the human-readable display texts
|
||
- `Concept.type`: the label of the resource the code belongs to (e.g. `Condition`)
|
||
|
||
```cypher
|
||
// CREATE CONCEPT NODES
|
||
|
||
MATCH (parent)-[]->(code:Code)
|
||
UNWIND keys(code) AS k
|
||
WITH parent, code, k
|
||
WHERE k ENDS WITH '_code'
|
||
|
||
// Extract values for coding prefix
|
||
WITH parent, code,
|
||
code[k] AS codeValue,
|
||
code[replace(k, '_code', '_system')] AS systemValue,
|
||
code[replace(k, '_code', '_display')] AS displayValue
|
||
WHERE codeValue IS NOT NULL AND systemValue IS NOT NULL
|
||
|
||
// Collect all coding pairs + displays
|
||
WITH parent, code,
|
||
collect(systemValue + '::' + codeValue) AS pairIds,
|
||
[d IN collect(displayValue) WHERE d IS NOT NULL] AS displays
|
||
|
||
// Create Concept IDs
|
||
WITH parent, code, displays,
|
||
apoc.text.join(apoc.coll.sort(pairIds), '||') AS conceptId,
|
||
labels(parent)[-1] AS conceptType
|
||
|
||
// Create shared Concept node
|
||
MERGE (c:Concept {id: conceptId})
|
||
SET c.displays = displays,
|
||
c.type = conceptType
|
||
|
||
// Connect Code to Concept
|
||
MERGE (code)-[:MAPS_TO]->(c);
|
||
```
|
||
|
||
### 3. Connect Patients to Concepts
|
||
|
||
Adds a direct `HAS_CONCEPT` shortcut from each patient to every concept found in their record, which makes cohort-style queries much simpler.
|
||
|
||
```cypher
|
||
// CONNECT PATIENT NODES TO CONCEPTS
|
||
|
||
MATCH (p:Patient)<-[]-(parent)-[]->(code:Code)-[:MAPS_TO]->(c:Concept)
|
||
MERGE (p)-[:HAS_CONCEPT]->(c);
|
||
```
|
||
|
||
**Example:** find the most frequent concepts across all patients:
|
||
```cypher
|
||
MATCH (p:Patient)-[:HAS_CONCEPT]->(c:Concept)
|
||
RETURN c.type AS type, c.displays AS displays, count(DISTINCT p) AS patients
|
||
ORDER BY patients DESC
|
||
LIMIT 25;
|
||
```
|
||
|
||
## ❓ Troubleshooting
|
||
|
||
**Common Issues:**
|
||
|
||
- **Connection refused to FHIR server**: Check your network settings and ensure the FHIR server is accessible from within the Docker container.
|
||
- **Authentication failures**: Verify your credentials in the `.env` file.
|
||
- **Container startup failures**: Ensure all required Docker ports are available and not used by other applications.
|
||
- **No data found in fhir bundle**: Ensure that the FHIR server is up and responding patient data. Try set the COMPLEX_PATIENTS variable to FALSE in your .env file. Some FHIR servers might not support the FHIR search logic.
|
||
- **`Unknown function 'apoc.text.join'` during post-processing**: The APOC plugin is not installed or enabled in your Neo4j instance.
|
||
|
||
## 📚 Architecture
|
||
|
||
The MeDaX pipeline consists of the following components:
|
||
|
||
1. **FHIR Client**: Connects to the FHIR server and retrieves patient data
|
||
2. **Data Transformer**: Converts FHIR resources into graph entities and relationships
|
||
3. **Reference Processor**: Converts references to relationships
|
||
4. **BioCypher Adapter**: Prepares the transformed data for Neo4j admin import
|
||
5. **Neo4j Database**: Stores and serves the graph representation of the health data
|
||
6. **Post-processing (optional)**: Cypher queries that add timelines and shared concept nodes
|
||
|
||
## ✍️ Citation
|
||
|
||
If you use the MeDaX pipeline in your research, please cite: 10.5281/zenodo.15229077 and Mazein, I and Gebhardt, T et al. [MeDaX, a knowledge graph on FHIR.](https://doi.org/10.3233/shti240423)
|
||
|
||
## 🙏 Acknowledgements
|
||
|
||
- We are leveraging [BioCypher](https://biocypher.org) [](https://doi.org/10.1038/s41587-023-01848-y) to create the Neo4j admin input.
|
||
- Remark: We introduced slight adjustments to BioCypher's code to support batching.
|
||
- We used BioCypher's git template as a starting point for our development:
|
||
- Lobentanzer, S., BioCypher Consortium, & Saez-Rodriguez, J. Democratizing knowledge representation with BioCypher [Computer software]. https://github.com/biocypher/biocypher
|
||
- We used synthetic data generated with [Synthea](https://doi.org/10.1093/jamia/ocx079) during the development process. This data is provided in the testData folder.
|
||
- We are using the [HAPI](https://hapifhir.io/) R4 server while developing and showcasing the capabilities of our tool.
|
||
- This project has been funded by the BMBF, FKZ: 01ZZ2019. |