Ignifyr (formerly toFHIR)

Everything that carried the tofhir name has been renamed to ignifyr. When upgrading an existing deployment:

The tofhir-redcap companion service and the srdc/tofhir-web UI image keep their current names until they are renamed in their own repositories; only Ignifyr's own configuration key for the integration changed (ignifyr-redcap).

Ignifyr is a powerful, FHIR-first ETL engine designed to map legacy health data to the HL7 FHIR standard. It is built to handle complex mappings with high performance and scalability.

It can be used as a library or a standalone tool for data integration. The standalone mode accepts command-line arguments to run batch executions or start a command-line interface (CLI) for interactive mapping management.

• Website: ignifyr.io
• Research Paper: To be published soon...

• Versatile Connectivity: Read from file systems, RDBMS, Apache Kafka, REDCap, or FHIR servers.
• Advanced Mapping: Utilizes the onfhir-template-engine to support 1-to-1, 1-to-many, many-to-1, and many-to-many mappings.
• Flexible Output: Generate HL7 FHIR resources and persist them to a FHIR endpoint (e.g., Repofyr), or write them to a file system as NDJSON, CSV, Parquet, or Delta Lake.

Ignifyr is a small core engine plus plugins. The engine reads a source, applies FHIR mappings, and hands the results to a sink; every concrete input and output is a plugin discovered at runtime through a Java ServiceLoader service-provider interface (IgnifyrExtension). The engine never names a connector, sink, format, or capability directly — it looks them up in its ExtensionRegistry by the settings type a job uses. The engine itself ships no concrete I/O at all: even the flagship FHIR-repository writer is a plugin. The practical upshot, and the core design rule, is: moving a feature between editions is a one-folder module move, with zero engine code changes.

The data flow is always the same: read source → apply mappings → write to sink → (optionally) archive the input. A mapping job's JSON parses everywhere, because every settings and model class lives in the engine regardless of which edition ships the corresponding plugin; if a job names a plugin that isn't installed, it fails at run time with an actionable "install …" message rather than a parse error.

The reactor splits into two editions. Community (Apache-2.0, published to Maven Central, shaded into the ignifyr-cli standalone jar) is the batch engine. Enterprise (private, shaded into the ignifyr-server jar) adds the REST server, streaming and scheduling, and the advanced connectors/formats. Each distribution's dependency list is the definition of its edition, so an edition change is one line moved between two POMs. A maven-enforcer gate (ban-enterprise-deps, opted into by the community modules) makes the boundary mechanical: spark-sql-kafka-0-10, cron4j, delta-spark, the DB2 JCC driver, and the Logstash encoder / Fluentd logger can never reach a community module or the community fat jar.

Each table lists what a module does and — since "why is this its own Maven module?" is the question the layout is designed to answer — the reason it is separate. The honest reasons fall into four kinds: dependency isolation (it carries a library the community edition must not ship), edition gating (the feature is paid, and the module boundary is the enforcement), cycle avoidance / layering (it cannot live where you would first put it), and seam anchoring (it is the extension point something else plugs into). Where a module exists mainly for structural regularity, the table says so.

The REST contract is ignifyr-server/api.yaml and the reference configuration is ignifyr-engine/src/main/resources/application.conf. Modules with non-obvious internals carry their own CLAUDE.md; run list-plugins on either jar to see what a given deployment actually has installed. How to build and test — including the short/long test tiers and the checks that keep the edition boundary honest — is in test-flow/README.md.

To run Ignifyr, you need:

• Java 11.0.2
• Scala 2.13
• An HL7 FHIR repository if you would like to persist the created resources (e.g., Repofyr)

Ignifyr can read data from the following data source types (the module providing each reader is in brackets):

And it can write the mapped results to:

Ignifyr can be utilized via the standalone Engine (CLI/Batch) or the Web Server (REST API).

When started as a standalone tool, the engine can run in two modes based on arguments:

• cli: Starts the interactive Command Line Interface (Default).
• run: Runs a configured mapping-job as a batch process and shuts down. run command accepts the following parameters:
  • --job: The path to the mapping-job to be executed. If provided, overrides the path provided to the JVM as the configuration parameter.
  • --mappings: The path to the mappings folder. If provided, overrides the path provided to the JVM as the configuration parameter.
  • --schemas: The path to the schemas folder. If provided, overrides the path provided to the JVM as the configuration parameter.
  • --db: The path to the database folder that is used for scheduled jobs. If provided, overrides the path provided to the JVM as the configuration parameter.
• extract-redcap-schemas: Extracts schemas from a REDCap data dictionary. Provided by the ignifyr-redcap module (Enterprise); without it, the command reports the module to install. extract-redcap-schemas command accepts the following parameters:
  • --data-dictionary: The path to the REDCap data dictionary
  • --definition-root-url: The root url of FHIR resources
  • --encoding: The encoding of CSV file whose default value is UTF-8 (OPTIONAL)

Once the interactive CLI is up, the following commands are available:

• help: Displays the help text and see the available commands and their use. Installed extension modules append their own commands to it.
• load <path>: Load the Mapping Job definition file from the path.
• reload: Reload the mapping definitions from their source into the mapping repository.
• run [<url>|<name>]: Run the task(s). Without a parameter, all task of the loaded Mapping Job are run. A specific task can be indicated with its name or URL. (Alias: execute.)
• list: Show jobs with at least one running mapping.
• list-plugins: Show the installed extension modules and everything they contribute — connectors, sinks, file formats, terminology/identity services, CLI commands, schema inferrers, and the streaming/scheduling capabilities. The quickest way to tell which edition/plugins a given jar carries.
• stop: Stop the execution of the Mapping Job (if any) or a specific Mapping Task associated with a job.
• exit: Exit the program. (Alias: quit.)

After the app is up and running, these commands are ready to be executed. If there is no mapping job loaded initially, firstly, a mapping job needs to be loaded with the command load <mapping-job-path>. This command loads the mapping job located in the path. After that, the mapping job can be run with the command run.

The server provides a REST API to manage the lifecycle of mapping projects.

• Base URL: http://:8085/ignifyr (default)
• API Documentation: SwaggerHub API Docs

Ignifyr uses HOCON-based configuration. Below is a snippet of the standard ignifyr.conf structure:

ignifyr {

  # A path to a directory from where any File system readings should use within the mappingjob definition.
  # This should be pointed to the root folder of the definitions.
  context-path = "ignifyr-definitions"

  mappings = {

    # The repository where the mapping definition are kept.
    repository = {
      folder-path = "mappings"
    }

    # Configuration of the schemas used in the mapping definitions.
    schemas = {
      repository = { # The repository where the schema definitions are kept.
        folder-path = "schemas"
      }
      # Specific FHIR version used for schemas in the schema repository.
      # Represents fhirVersion field in the standard StructureDefinition e.g. 4.0.1, 5.0.0
      fhir-version = "4.0.1"
    }

    contexts = {
      # The repository where the context definitions are kept.
      repository = {
        folder-path = "mapping-contexts"
      }
    }

    # Timeout for each mapping execution on an individual input record
    timeout = 5 seconds
  }

  mapping-jobs = {
    repository = { # The repository where the job definitions are kept.
      folder-path = "mapping-jobs"
    }
    # Absolute path to the JSON file for the MappingJob definition to load at the beginning
    # initial-job-file-path = "mapping-jobs/project1-mappingjob.json"

    # Number of partitions to repartition the source data before executing the mappings for the mapping jobs
    # numOfPartitions = 10

    # Maximum number of records for batch mapping execution, if source data exceeds this it is divided into chunks
    # maxChunkSize = 10000
  }

  terminology-systems = {
    # The path to the folder where Terminology System files (config files, CodeSystems, ConceptMaps etc.) are kept.
    folder-path = "terminology-systems"
  }

  archiving = {
    # Folder to keep erroneous records
    erroneous-records-folder = "erroneous-records-folder"
    
    # Folder to keep archived files
    archive-folder = "archive-folder"
    
    # Frequency in milliseconds to run the archiving task for file streaming jobs
    stream-archiving-frequency = 5000
  }

  # Settings for FHIR repository writer
  fhir-server-writer {
    # The # of FHIR resources in the group while executing (create/update) a FHIR batch operation.
    batch-group-size = 50
  }

  # Database folder of Ignifyr (e.g., to maintain synchronization times for scheduled jobs)
  db-path = "ignifyr-db"
}

# Spark configurations
spark = {
  app.name = "DataTools4Heart Data Integration Suite"
  master = "local[4]"
  # Directory to store Spark checkpoints
  checkpoint-dir = "checkpoint"
}

akka = {
  daemonic = "on"
}

Considering the configuration file defined above, Ignifyr can be utilized with a folder structure like the following:

ignifyr-definitions (root folder of definitions)
├── mappings
│   ├── project1
│   │   ├── mapping1.json
│   │   ├── mapping2.json
│   │   └── ...
├── mapping-jobs
│   ├── project1
│   │   ├── mapping-job1.json
│   │   ├── mapping-job2.json
│   │   └── ...
├── schemas
│   ├── project1
│   │   ├── schema1.json
│   │   ├── schema2.json
│   │   └── ...
├── mapping-contexts
│   ├── project1
│   │   ├── context1.json
│   │   ├── context2.json
│   │   └── ...
├── terminology-systems
│   ├── terminology1
│   ├── ├── ConceptMap1.csv
│   ├── ├── CodeSystem1.csv
│   └── ...
└── ignifyr.conf

You are free to organize the definitions in any way you like and arrange the configuration file accordingly. However, we are suggesting to keep the definitions in a folder structure as shown above to keep the definitions organized and easy to manage.

A container for schemas, mappings, mapping contexts, and mapping jobs. It organizes definitions into logical groups.

Please note that, terminology systems are not included in a project. They are defined separately and can be used in any project.

A schema is a definition of the structure of the source data. It is used to validate the source data and to provide the context for the mappings. They are nothing but the simple HL7 FHIR StructureDefinition resources that are defined in JSON format.

An example of a simple mapping definition file:

{
  "url": "https://aiccelerate.eu/fhir/mappings/project1/patient-mapping",
  "name": "patient-mapping",
  "title": "Mapping of patient data to Patient FHIR Resource",
  "source": [{
    "alias": "patient",
    "url": "https://aiccelerate.eu/fhir/schemas/project1/patient"
  }],
  "mapping": [
    {
      "expression": {
        "name": "result",
        "language": "application/fhir-template+json",
        "value": {
          "resourceType": "Patient",
          "id": "{{mpp:getHashedId('Patient',pid)}}",
          "meta": {
            "source": "{{%sourceSystem.sourceUri}}"
          },
          "active": true,
          "identifier": [
            {
              "use": "official",
              "system": "{{%sourceSystem.sourceUri}}",
              "value": "{{pid}}"
            }
          ],
          "gender": "{{gender}}",
          "birthDate": "{{birthDate}}",
          "deceasedDateTime": "{{? deceasedDateTime}}",
          "address": {
            "{{#pc}}": "{{homePostalCode}}",
            "{{?}}": [
              {
                "use": "home",
                "type": "both",
                "postalCode": "{{%pc}}"
              }
            ]
          }
        }
      }
    }
  ]
}

The json snippet above illustrates the structure of an example mapping. On the top, the url, name, and title fields are the metadata of the mapping. The source field is used to define the source schema of the mapping. The mapping field is the list of mapping definitions. The real magic in mappings happens in the expression fields (e.g. {{<expression>}} ). Ignifyr uses the expression to generate the FHIR resources by using onfhir-template-engine. By doing so, it can generate the FHIR resources based on the source data.

For example, considering {{gender}} expression, it refers to "gender" column in the source data. When this mapping is executed, each record at "gender" column in the source replaces the expression and generate the FHIR resources.

The json keys in the expression.value represent the FHIR resource attributes. That is, we write the FHIR resource structure by providing the values through a template language where we can access the fields of the source data as defined by its schema. On the value sides, onfhir-template-engine is used to interpret the source data. You can get more information how template engine works on the GitHub page.

Mapping contexts are CSV files that have a specific format. It refers to the additional information utilized when defining a mapping. Mapping contexts facilitates easy exchange and integration between healthcare concepts and referenced within the mappings.

They provide two functionalities:

• Concept Map: Enables the mapping of different health-care concept codes between different systems.
• Unit Conversion: Enables mappings between different healthcare data units. e.g. mg → g, cm → m, etc.

Let's say that the data source has its own EHR codes for the concepts, and we want to map these codes to the ICD10-PCS codes. In this case, we can define and use a concept map file to define the mappings between the source codes and the ICD10-PCS codes:

Assume, we have a mapping context file named heart-transplantation-code-map.csv and it is located in the mapping-contexts/project1 folder. Then, we can use this context file in the mapping definitions as follows:

{
  "id" : "procedure-mapping",
  "url" : "https://aiccelerate.eu/mappings/amc/procedure-mapping",
  "name" : "procedure-mapping",
  "title" : "Mapping of EHR table to Procedure FHIR Resource",
  "source" : [ {
    "alias" : "Procedure",
    "url" : "https://aiccelerate.eu/schemas/project1/procedure",
    "joinOn" : [ ]
  } ],
  "context" : {
    "heartTransplantationCodeMap" : {
      "category" : "concept-map",
      "url" : "$CONTEXT_REPO/project1/heart-transplantation-code-map.csv"
    }
  },
  "variable" : [ ],
  "mapping" : [ {
    "expression" : {
      "name" : "result",
      "language" : "application/fhir-template+json",
      "value" : {
        "resourceType" : "Procedure",
        "category" : [ {
          "coding" : [ {
            "system" : "http://snomed.info/sct",
            "code" : "387713003",
            "display" : "Surgical procedure (procedure)"
          } ]
        } ],
        "code": {
          "coding": [
            {
              "system": "http://hl7.org/fhir/sid/icd-10-pcs",
              "code": "{{ mpp:getConcept(%heartTransplantationCodeMap, type_code, 'target_code') }}",
              "display": "{{ mpp:getConcept(%heartTransplantationCodeMap, type_code, 'target_display') }}"
            }
          ]
        }
      }
    }
  } ]
}

Firstly, context file is registered in the mapping definition with the context field and a name is given to it (e.g. heartTransplantationCodeMap). Then, the mpp:getConcept function is used to get the target code and display name from the context file. The first parameter of the function is the name given for the context file, the second parameter is the source code, and the third parameter is the target field name. When the mapping is executed, type_code is replaced with the actual source code and the function returns the target code and display name. For example, if the source code is xyz, the function returns 02YA0Z0 for the code and Transplantation of Heart, Allogeneic, Open Approach for the display name.

Another use case for mapping contexts is the unit conversion. Let's say that the source data has the lab results in different units, and we want to convert them. In this case, we can define and use a unit conversion file to define the conversion between units:

Similarly, assume, we have a mapping context file named lab-unit-conversion.csv and it is located in the mapping-contexts/project1 folder. Then, we can use this context file in the mapping definitions as follows:

{
  "id" : "lab-mapping",
  "url" : "https://aiccelerate.eu/mappings/amc/lab-mapping",
  "name" : "lab-mapping",
  "title" : "Mapping of EHR table to Observation FHIR Resource",
  "source" : [ {
    "alias" : "Lab",
    "url" : "https://aiccelerate.eu/schemas/project1/lab",
    "joinOn" : [ ]
  } ],
  "context" : {
    "labUnitConversion" : {
      "category" : "unit-conversion",
      "url" : "$CONTEXT_REPO/project1/lab-unit-conversion.csv"
    }
  },
  "variable" : [ ],
  "mapping" : [ {
    "expression" : {
      "name" : "result",
      "language" : "application/fhir-template+json",
      "value" : {
        "resourceType" : "Observation",
        "code" : {
          "coding" : [ {
            "system" : "http://loinc.org",
            "code" : "789-8",
            "display" : "Erythrocytes [#/volume] in Blood by Automated count"
          } ]
        },
        "valueQuantity" :"{{ mpp:convertAndReturnQuantity(%labResultUnitConversion, lab_code, value, unit) }}"
      }
    }
  } ]
}

Similarly, context file is registered in the mapping definition with the context field and a name is given to it (e.g. labUnitConversion). Then, the mpp:convertAndReturnQuantity function is used to convert the lab result value to the target unit. The first parameter of the function is the name of the context file, the second parameter is the lab code (e.g. 5060), the third parameter is the measured value for that lab, and the fourth parameter is the source unit (e.g. g/L). When the mapping is executed, the function converts the value to the target unit by applying the conversion function and returns the converted value. For example, if the lab_code is 5060, the value is 5, and the unit is g/L, the function calculates 5000 for the value in the mg/L unit and returns a HL7 FHIR Quantity object:

{
  "value": 5000,
  "unit": "mg/L",
  "system": "http://unitsofmeasure.org",
  "code": "mg/L"
}

If not set explicitly, Ignifyr uses the batch mode by default. In the batch mode, Ignifyr goes through these steps:

1. Reads the source data
2. Executes the mappings
3. Persists the generated FHIR resources to the sink
4. Optionally, archives the source data and save erroneous records
5. Exits

This means that your source data is expected to be a static file/table or a set of files/tables that are not expected to be updated during the execution of the mapping job. Example of a Mapping Job definition file with csv source type:

{
  "id": "project1-mapping-job",
  "sourceSettings": {
    "source": {
      "jsonClass": "FileSystemSourceSettings",
      "name": "project1-source",
      "sourceUri": "https://aiccelerate.eu/fhir/data-integration-suite/project1-data",
      "dataFolderPath": "test-data/project1"
    }
  },
  "sinkSettings": {
    "jsonClass": "FhirRepositorySinkSettings",
    "fhirRepoUrl": "http://localhost:8081/fhir"
  },
  "dataProcessingSettings": {
    "saveErroneousRecords": false,
    "archiveMode": "off"
  },
  "mappings": [
    {
      "name": "patient-mapping",
      "mappingRef": "https://aiccelerate.eu/fhir/mappings/project1/patient-mapping",
      "sourceBinding": {
        "patient": {
          "jsonClass": "FileSystemSource",
          "path": "patients.csv",
          "contentType": "csv"
        }
      }
    },
    {
      "name": "practitioner-mapping",
      "mappingRef": "https://aiccelerate.eu/fhir/mappings/project1/practitioner-mapping",
      "sourceBinding": {
        "practitioner": {
          "jsonClass": "FileSystemSource",
          "path": "practitioners.csv",
          "contentType": "csv"
        }
      }
    }
  ]
}

The json snippet above illustrates the structure of an example mapping job. Let's go through the fields one by one:

• sourceSettings defines the source settings of the mapping job. The source settings config is used to connect to the source data. In this case, the source type of data is file system source and dataFolderPath defines the path of the source data folder. Please note that, dataFolderPath is a relative path to the root folder of the definitions. Also, it may be an absolute path as well.
• Assuming onFHIR is running on the system, sinkSettings defines FHIR endpoint configurations to connect to the data destination.
• dataProcessingSettings is used to define post-processes after the mapping is completed. It is explained in detail here: Data Processing Settings
• mappings is a list of mapping tasks that mapping job includes. For a purpose of illustration, the mapping job above includes two mappings:
  • https://aiccelerate.eu/fhir/mappings/project1/patient-mapping
  • https://aiccelerate.eu/fhir/mappings/project1/practitioner-mapping

Let's take the patient mapping as an example from the mappings list. https://aiccelerate.eu/fhir/mappings/project1/patient-mapping is the unique reference URL of the mapping repository. Assuming this URL refers to the first mapping example in the mapping section: patient-mapping, this means that patient mapping will be executed with the source data defined in the sourceBinding part. Inside sourceBinding part, patient is the alias of the source data, and it should match with the alias used in source field in the mapping.

jsonClass specifies the type of the source, and path is the file name of the source data. Since we have FileSystemSourceSettings defined in the source settings, jsonClasses of mappings are expected to be FileSystemSource. For the file source mappings, the path field should be specified, and it represents the data source file of each mapping. This field is a relative path to the dataFolderPath defined in the source settings.

Ignifyr supports streaming of file system in case you want to continuously monitor the changes on the source data and stream the newcoming/updated data to Ignifyr mapping executions. This can be done with the asStream config parameter of the source. If it is set to true, Ignifyr will monitor the FileSystemSource files defined at path paths and trigger the mapping executions in case the files are updated. Ignifyr automatically marks the processed data source files and only processes the newcoming/updated records.

Ignifyr goes through these steps in the streaming mode:

1. Reads the initial existing source data
2. Executes the mappings
3. Persists the generated FHIR resources to the sink
4. Optionally, archives the source data and save erroneous records
5. Monitors the source data for changes
6. Executes the mappings for the newcoming/updated data
7. Persists the generated FHIR resources to the sink
8. Optionally, archives the source data and save erroneous records
9. Repeats the steps 5-8

Example of a Mapping Job definition file with csv source type in streaming mode:

{
  "id": "project1-mapping-job",
  "sourceSettings": {
    "source": {
      "jsonClass": "FileSystemSourceSettings",
      "name": "project1-source",
      "sourceUri": "https://aiccelerate.eu/data-integration-suite/project1-data",
      "dataFolderPath": "D:/ignifyr/data",
      "asStream": true
    }
  },
  "sinkSettings": {
    "jsonClass": "FhirRepositorySinkSettings",
    "fhirRepoUrl": "http://localhost:8081/fhir"
  },
  "dataProcessingSettings": {
    "saveErroneousRecords": false,
    "archiveMode": "off"
  },
  "mappings": [
    {
      "name": "patient-mapping",
      "mappingRef": "https://aiccelerate.eu/fhir/mappings/project1/patient-mapping",
      "sourceBinding": {
        "patient": {
          "jsonClass": "FileSystemSource",
          "path": "patients",
          "contentType": "csv"
        }
      }
    }
  ]
}

The json snippet above illustrates the structure of an example mapping job in streaming mode. Similar to the batch mode, most of the fields are the same. The only differences are:

• asStream field in the source settings
• path in the source binding of the mapping. path should be the name of the folder this time, and it is where Ignifyr will monitor the changes.

Similarly, if we had a source with SQL type, sourceSettings and mappings part would look like this:

{
  "sourceSettings": {
    "source": {
      "jsonClass": "SqlSourceSettings",
      "name": "project1-source",
      "sourceUri": "https://aiccelerate.eu/data-integration-suite/project1-data",
      "dataFolderPath": "jdbc:postgresql://localhost:5432/db_name",
      "username": "postgres",
      "password": "postgres"
    }
  }
}

{
  "name": "location-sql-mapping",
  "mappingRef": "https://aiccelerate.eu/fhir/mappings/location-sql-mapping",
  "sourceBinding": {
    "source": {
      "jsonClass": "SqlSource",
      "tableName": "location"
    }
  }
}

We can give a table name with the tableName field, as well as write a query with the query field:

{
  "name": "location-sql-mapping",
  "mappingRef": "https://aiccelerate.eu/fhir/mappings/location-sql-mapping",
  "sourceBinding": {
    "source": {
      "jsonClass": "SqlSource",
      "query": "select * from location"
    }
  }
}

Mapping job and mapping examples shown below for the streaming type of sources like Kafka:

{
  "sourceSettings": {
    "source": {
      "jsonClass": "KafkaSourceSettings",
      "name": "project1-source",
      "sourceUri": "https://aiccelerate.eu/data-integration-suite/project1-data",
      "bootstrapServers": "localhost:9092,localhost:9093"
    }
  }
}

{
  "name": "location-sql-mapping",
  "mappingRef": "https://aiccelerate.eu/fhir/mappings/location-sql-mapping",
  "sourceBinding": {
    "source": {
      "jsonClass": "KafkaSource",
      "topicName": "patients",
      "options": {
        "startingOffsets": "earliest"
      }
    }
  }
}

Ignifyr only considers the value field of kafka topics. Therefore, when you subscribe a topic, Ignifyr waits for string-type data but in correct JSON format. For example, when you want to use the data in the topic, you should publish the data in the following format:

{
  "pid": "p1",
  "gender": "male",
  "birthDate": "1995-11-10"
}

Ignifyr integrates seamlessly with RedCAP through the tofhir-redcap integration service, which publishes RedCAP records to Kafka. Utilize the same configuration approach as described for Kafka, with a few key considerations:

• Source Configuration: In the mapping job's sourceSettings, use a plain Kafka source configuration. Environment variables (e.g. ${REDCAP_PROJECT_ID}) referenced in sourceUri or in the topic names are resolved automatically. Here's an example JSON configuration:

{
  "sourceSettings": {
    "source": {
      "jsonClass": "KafkaSourceSettings",
      "name": "project1-source",
      "sourceUri": "https://aiccelerate.eu/data-integration-suite/project1-data",
      "bootstrapServers": "localhost:9092,localhost:9093"
    }
  }
}

• Topic Name: While defining the topic name for a mapping source binding within a mapping job, use the name generated by the tofhir-redcap integration service for the corresponding RedCAP project. For detailed instructions, refer to the README file of the integration service.

Below is an example configuration for mapping jobs using a FHIR Server data source:

{
  "sourceSettings" : {
    "source" : {
      "jsonClass" : "FhirServerSourceSettings",
      "name" : "pilot1-source",
      "sourceUri" : "https://aiccelerate.eu/data-integration-suite/pilot1-data",
      "serverUrl" : "http://localhost:8082",
      "securitySettings": {
        "jsonClass": "BasicAuthenticationSettings",
        "username": "username",
        "password": "password"
      }
    }
  }
}

In addition to specifying the server URL (serverUrl), you can configure security settings via the securitySettings field.

Within the mapping source, you can define the resource type (e.g., Patient, Observation) and apply filters using a query string:

{
  "name": "patient-mapping",
  "mappingRef" : "https://aiccelerate.eu/fhir/mappings/pilot1/patient-mapping",
  "sourceBinding" : {
    "source" : {
      "jsonClass" : "FhirServerSource",
      "resourceType" : "Patient",
      "query": "gender=male&birtdate=ge1970"
    }
  }
}

Since Ignifyr uses Apache Spark in its core, you can give any option that is supported by Apache Spark. Available options for different source types can be found in the following links:

• File System
  • CSV & TSV: https://spark.apache.org/docs/3.4.1/sql-data-sources-csv.html#data-source-option
  • JSON: https://spark.apache.org/docs/3.4.1/sql-data-sources-json.html#data-source-option
  • Parquet: https://spark.apache.org/docs/3.4.1/sql-data-sources-parquet.html#data-source-option
• SQL: https://spark.apache.org/docs/3.4.1/sql-data-sources-jdbc.html#data-source-option
• Apache Kafka: https://spark.apache.org/docs/3.4.1/structured-streaming-kafka-integration.html

To give any spark option, you can use the options field in the source binding of the mapping in a mapping job.

{
  "name": "patient-mapping",
  "mappingRef": "https://aiccelerate.eu/fhir/mappings/project1/patient-mapping",
  "sourceBinding": {
    "source": {
      "jsonClass": "FileSystemSource",
      "path": "patients",
      "contentType": "csv",
      "options": {
        "sep": "\\t" // tab separated file
      }
    }
  }
}

In a mapping job, you can read data from more than one data source. Let's consider a scenario where you have different sources

• patient-test-data: Contains patient identifiers, specifically the "pid" column and this information is in a CSV file named patient.csv under /test-data folder.
• patient-gender-test-data: Contains gender information for patients, including "pid" and "gender" columns and this information is served from a Postgres database.

We'll implement a mapping job that utilizes these two CSV files as data sources and runs a simple patient mapping.

First, define the source settings pointing to the two different data sources:

{
  "sourceSettings" : {
    "patientSource" : {
      "jsonClass" : "FileSystemSourceSettings",
      "name" : "patient-test-data",
      "sourceUri" : "http://test-data",
      "dataFolderPath" : "/test-data",
      "asStream" : false
    },
    "genderSource" : {
      "jsonClass" : "SqlSourceSettings",
      "name" : "patient-gender-test-data",
      "sourceUri" : "http://test-data-gender",
      "databaseUrl" : "jdbc:postgresql://localhost:5432/test-data-gender",
      "username" : "user",
      "password" : "pass"
    }
  }
}

The patientSource points to the test-data directory in the file system, while the genderSource points to a relational database, actually a query result or a table name. It is important to note that the mapping definitions are not directly connected to the data sources. genderSource can point to a folder which means that the same mapping can be executed on the data read from different sources.

{
  "sourceSettings" : {
    "patientSource" : {
      "jsonClass" : "FileSystemSourceSettings",
      "name" : "patient-test-data",
      "sourceUri" : "http://test-data",
      "dataFolderPath" : "/test-data",
      "asStream" : false
    },
    "genderSource" : {
      "jsonClass" : "FileSystemSourceSettings",
      "name" : "patient-gender-test-data",
      "sourceUri" : "http://test-data-gender",
      "dataFolderPath" : "/test-data-gender",
      "asStream" : false
    }
  }
}

Next, specify the source bindings for your mappings in the job. Here's an example:

{
  "mappings" : [ {
    "name": "patient-mapping-with-two-sources",
    "mappingRef" : "http://patient-mapping-with-two-sources",
    "sourceBinding" : {
      "patient" : {
        "jsonClass" : "FileSystemSource",
        "path" : "patient-simple.csv",
        "contentType" : "csv",
        "options" : { },
        "sourceRef": "patientSource"
      },
      "patientGender" : {
        "jsonClass" : "SqlSource",
        "query" : "SELECT pid, gender FROM patient_gender",
        "sourceRef": "genderSource"
      }
    }
  } ]
}

In this example, patient-simple.csv is used for the patient mapping source, while an SQL query result is used for the patientGender mapping source. Since the mapping job has more than one data source, we should specify the source reference in the mapping source binding using sourceRef field. Here, patient source reads the csv file from patientSource whereas patientGender source reads the result of an SQL query from genderSource.

If sourceRef is skipped or does not match any entry in the sourceSettings, the first source specified in sourceSettings will be used to read the data.

If the genderSource was connected to file system in the job definition, the sourceBinding parameters would be as in the following:

{
  "mappings" : [ {
    "name": "patient-mapping-with-two-sources",
    "mappingRef" : "http://patient-mapping-with-two-sources",
    "sourceBinding" : {
      "patient" : {
        "jsonClass" : "FileSystemSource",
        "path" : "patient-simple.csv",
        "contentType" : "csv",
        "options" : { },
        "sourceRef": "patientSource"
      },
      "patientGender" : {
        "jsonClass" : "FileSystemSource",
        "path" : "patient-gender-simple.csv",
        "contentType" : "csv",
        "options" : { },
        "sourceRef": "genderSource"
      }
    }
  } ]
}

Finally, in the mapping definition, join these two data sources:

{
  "source": [
    {
      "alias": "patient",
      "url": "http://patient-schema",
      "joinOn": [
        "pid"
      ]
    },
    {
      "alias": "patientGender",
      "url": "http://patient-gender",
      "joinOn": [
        "pid"
      ]
    }
  ]
}

Specify the corresponding schema URL for each data source. Use the same source keys (patient and patientGender) as alias to match schemas with the data sources provided in the mapping job definition. Then, join the two source data using the pid column available in both.

The first source i.e. patient is called the main schema, and its fields are accessible directly in the mapping. To access attributes of other schemas (side schemas), use the % operator (e.g., %patientGender).

Here's an example mapping that utilizes the pid field from the patient source and the gender information from the patientGender source:

{
  "gender": "{{%patientGender.gender}}",
  "id": "{{pid}}"
}

Please refer to the following files for full definitions:

• patient-simple.csv
• patient-gender-simple.csv
• patient-mapping-job-with-two-sources.json
• patient-mapping-with-two-sources.json

Ignifyr supports persisting the generated FHIR resources to a FHIR repository. The sink settings are defined in the mapping job definition file. The following example shows the sink settings for a FHIR repository:

{
  "sinkSettings": {
    "jsonClass": "FhirRepositorySinkSettings",
    "fhirRepoUrl": "http://localhost:8081/fhir"
  }
}

Or you can use a local file system to persist the generated FHIR resources:

{
  "sinkSettings": {
    "jsonClass": "FileSystemSinkSettings",
    "path": "sink/project1",
    "contentType": "csv"
  }
}

Each sink is a plugin module, and the file sink's output formats are pluggable in turn:

A contentType whose handler is not installed fails on the first write with a message naming the module to install; the job itself still parses.

A FHIR terminology service can be automatically used by Ignifyr to handle concept lookup and concept map operations. If a terminology service is configured, mapping definitions can use lookup and translation services for codes/values of codesystems/valuesets.

An available FHIR terminology service can be configured as in the following:

...
  "terminologyServiceSettings": {
    "jsonClass": "FhirRepositorySinkSettings",
    "fhirRepoUrl": "https://fhir.loinc.org/",
    "securitySettings":{
        "jsonClass": "BasicAuthenticationSettings",
        "username": "???",
        "password": "???"       
    }   
  }
...

Ignifyr provides a LocalFhirTerminologyService which allows to use text files for concept details and translations. You can provide the concept map files or code/codesystem details by configuring the terminology service as in the following example:

...
  "terminologyServiceSettings": {
    "jsonClass": "LocalFhirTerminologyServiceSettings",
    "folderPath": "./src/test/resources/terminology-service",
    "conceptMapFiles": [
      {
        "fileName": "sample-concept-map.csv",
        "conceptMapUrl": "http://example.com/fhir/ConceptMap/sample1",
        "sourceValueSet": "http://terminology.hl7.org/ValueSet/v2-0487",
        "targetValueSet": "http://snomed.info/sct?fhir_vs"
      }
    ],
    "codeSystemFiles": [
      {
        "fileName":"sample-code-system.csv",
        "codeSystem": "http://snomed.info/sct"
      }
    ]
  }
...

Ignifyr's FHIRPath engine provides two functions becoming available when a terminology service is configured:

• trms:lookupDisplay: Lookup the display name of a given code and code system
• trms:translateToCoding: Translate the give code+codesystem within a valueset to the target code+codesystem (formatted as Coding) within target valueset.

The following example gets the display name in German (de column) of the code 119323008 defined in SNOMED code system:

{
  "system": "http://snomed.info/sct",
  "code": "111",
  "display": "{{ trms:lookupDisplay('119323008','http://snomed.info/sct','de') }}"
}

Similarly, when you want to translate the given code+system according to the given source value set and (optional) target value set, you can do something like this. This creates a FHIR-Coding object automatically and replaces the expression.

{
  "coding": [
    "{{? trms:translateToCoding(type,'http://terminology.hl7.org/CodeSystem/v2-0487','http://terminology.hl7.org/ValueSet/v2-0487', 'http://snomed.info/sct?fhir_vs')}}"
  ]
}

Ignifyr allows you to use a FHIR endpoint as and identity service in case FHIR resource identifiers need to be fetched given the business identifiers. In this case, you can use the idxs:resolveIdentifier function with the following parameters: idxs:resolveIdentifier(FHIR resource type, Identifier.value, Identifier.system) which returns a FHIR reference such as Patient/455435464698.

The following example puts the FHIR resource id of the Patient into the reference field by using the identity service:

{
  "subject": {
    "reference": "{{idxs:resolveIdentifier('Patient', pid, 'https://aiccelerate.eu/data-integration-suite/test-data')}}"
  }
}

Ignifyr supports running scheduled jobs with defined time ranges. To do so, you need to specify a cron expression in the mapping job definitions. Ignifyr uses cron4j library to handle scheduled jobs. Scheduled patterns for the expression can be found in the documentation section of cron4j. Synchronization times for scheduled jobs are maintained in a folder defined db-path setting in the configuration file.

You can schedule a mapping job as follows:

mapping-job.json

{
  ...
  "schedulingSettings": {
    "jsonClass": "SchedulingSettings",
    "cronExpression": "59 11 * * *"
  },
  ...
}

59 11 * * * pattern causes a task to be launched at 11:59AM every day.

Moreover, if your data source is SQL-based and contains time or date information, and you want to pull data at time intervals according to schedule, you can specify the initial time in your mapping job definition as follows:

mapping-job.json

{
  ...
  "schedulingSettings": {
    "jsonClass": "SQLSchedulingSettings",
    "cronExpression": "59 11 * * *",
    "initialTime": "2000-01-01T00:00:00"
  },
  ...
}

mapping.json

{
  ...,
  "name": "procedure-occurrence-mapping",
  "mappingRef": "https://aiccelerate.eu/fhir/mappings/omop/procedure-occurrence-mapping",
  "sourceBinding": {
    "source": {
      "jsonClass": "SqlSource",
      "query": "select ... from procedure_occurrence po left join concept c on po.procedure_concept_id = c.concept_id where po.procedure_date > $fromTs and po.procedure_date < $toTs"
    }
  },
  ...
}

procedure_occurrence table has a date column procedure_date in this example. When your scheduled task runs, $fromTs and $toTs placeholders are replaced with corresponding timestamps. According to the mapping job and mapping shown above, after you run the mapping job, lets say at 2022-08-08T10:05:30, the following variables will take place as the scheduled job runs.

Ignifyr supports archiving of erroneous records and the source data files. If you want to archive only the erroneous records, which are the records that could not be processed/mapped by the mapping job, you can specify the config in the mapping job definitions. The erroneous records are saved in the erroneous-records-folder defined in the sub-config of the archiving config in the configuration file.

mapping-job.json

{
  ...
  "dataProcessingSettings": {
    "saveErroneousRecords": true
  },
  ...
}

If you want to archive the source data files after processing, regardless of whether that file was processed/mapped successfully or not, you can specify the config in the mapping job definitions. The source data files are saved in the archive-folder defined in the sub-config of the archiving config in the configuration file.

mapping-job.json

{
  ...
  "dataProcessingSettings": {
    "archiveMode": "archive"
  },
  ...
}

Or if you want to simply delete the source data files after processing/mapping:

mapping-job.json

{
  ...
  "dataProcessingSettings": {
    "archiveMode": "delete"
  },
  ...
}

Or both. This will delete the source files after processing/mapping and save the erroneous records:

mapping-job.json

{
  ...
  "dataProcessingSettings": {
    "saveErroneousRecords": true,
    "archiveMode": "delete"
  },
  ...
}

While archiveMode works on a file-based basis, saveErroneousRecords works for each record/row in the source data.

Please also note that, the archiveMode config is only applicable for the file system source type.

When dealing with large data sources (e.g., 10 million+ rows), loading all data into memory at once is not practical. Ignifyr supports a batching strategy that allows you to process data in smaller batches by defining parameter sets that filter the source data.

Each batch is processed sequentially: data is loaded, mapped, written to the sink, and then memory is freed before moving to the next batch.

Define a batchingStrategy in your FhirMappingTask along with a preprocessSql that uses parameter placeholders (prefixed with $):

{
  "name": "measurement-mapping",
  "mappingRef": "https://example.com/measurement-mapping",
  "sourceBinding": {
    "source": {
      "jsonClass": "SqlSource",
      "query": "SELECT * FROM MEASUREMENT",
      "preprocessSql": "SELECT * FROM MEASUREMENT WHERE EXTRACT(YEAR FROM MEASUREMENT_DATE) = $year"
    }
  },
  "batchingStrategy": {
    "batchParameterSets": [
      {"year": "2018"},
      {"year": "2019"},
      {"year": "2020"}
    ]
  }
}

In this example, Ignifyr will execute 3 batches, one for each year. The $year placeholder in preprocessSql is replaced with values from batchParameterSets.

You can use multiple parameters for finer-grained batching:

{
  "name": "encounter-mapping",
  "mappingRef": "https://example.com/encounter-mapping",
  "sourceBinding": {
    "encounterMain": {
      "jsonClass": "FileSystemSource",
      "path": "encounters.csv",
      "contentType": "csv",
      "preprocessSql": "SELECT * FROM encounterMain WHERE EXTRACT(YEAR FROM ENCOUNTER_DATE) = $year AND EXTRACT(MONTH FROM ENCOUNTER_DATE) = $month"
    }
  },
  "batchingStrategy": {
    "batchParameterSets": [
      {"year": "2020", "month": "01"},
      {"year": "2020", "month": "02"},
      {"year": "2020", "month": "03"}
    ]
  }
}

ID Ranges with Custom Logic:

{
  "preprocessSql": "SELECT * FROM PATIENTS WHERE patient_id BETWEEN $id_start AND $id_end",
  "batchingStrategy": {
    "batchParameterSets": [
      {"id_start": "1", "id_end": "100000"},
      {"id_start": "100001", "id_end": "200000"},
      {"id_start": "200001", "id_end": "300000"},
      {"id_start": "300001", "id_end": "400000"}
    ]
  }
}

Key points:

• Parameter placeholders in preprocessSql use the $parameterName syntax
• All parameters in the placeholder must exist in each object of batchParameterSets
• Batches are processed sequentially, reducing memory footprint
• Only sources with preprocessSql containing parameter placeholders are affected by the batching strategy

For large ranges (e.g., 40 years × 12 months), you can generate batchParameterSets programmatically using the helper scripts:

import json

def generate_year_month_batches(start_year, end_year):
    """Generate batch parameter sets for year-month combinations"""
    batches = []
    for year in range(start_year, end_year + 1):
        for month in range(1, 13):
            batches.append({
                "year": str(year),
                "month": str(month)
            })
    return batches

# Generate batches for 1980-2020 (40 years)
batches = generate_year_month_batches(1980, 2020)

print(f"Total batches: {len(batches)}")  # Output: 492 batches

# Create the batching strategy
batching_strategy = {
    "batchParameterSets": batches
}

# Save to file
with open('batching_strategy.json', 'w') as f:
    json.dump(batching_strategy, f, indent=2)

print("Batching strategy saved to batching_strategy.json")