Validating NHS RTDS Data in Apache NiFi 2 with Avro Schemas

If you’re building a NiFi 2 pipeline to ingest RTDS CSV exports, you need to answer one question at the point of ingestion: is this row structurally valid before it goes anywhere near a database, a transformation, or a downstream API? The answer is a schema-validated CSVReader. But getting there on NiFi 2 with Avro is full of sharp edges. This article walks through all of them — for both the Attendance and Prescription feeds.

The two RTDS CSV feeds

RTDS submissions arrive as two structurally distinct CSV files. They share a patient identity spine but carry very different clinical payloads.

Attendance records a single radiotherapy treatment fraction — one visit, one set of procedures delivered. It is high-frequency, operationally lean, and built around demographic identity and procedure codes.

Prescription records the clinical intent behind a course of treatment — the diagnosis, the plan, the prescribed dose, the beam configuration. It is richer, more complex, and sits upstream of attendance in the patient journey.

Both files have fixed headers that will not change between extracts. That stability matters: it means we can rely on positional column mapping when alias resolution is uncertain.

Why Avro schema validation fails out of the box

Two failure modes will catch you before you read a single row.

Illegal field names. Avro field names must match [A-Za-z_][A-Za-z0-9_]*. Every RTDS column header violates this — spaces, parentheses, and mixed case are all illegal. NiFi rejects the schema immediately at controller service validation:

Not a valid Avro Schema: Illegal character in: NHS NUMBER

Missing defaults on nullable unions. When a field is optional, the natural instinct is ["null", "string"]. But Avro requires that when null is the first type in a union, a "default": null must be explicitly declared. Without it, NiFi rejects the entire schema before reading a single row — a controller-level error, not a row-level one, which can be disorienting to debug.

The fix for both is the same: use valid snake_case Avro field names, map back to the original CSV headers via the "aliases" property, and always pair nullable unions with "default": null.

The three rules

These apply to every NiFi Avro schema against real-world data.

1. Avro names are identifiers, not labels. If your source uses human-readable column headers — anything from a national specification or a legacy system — assume they contain illegal characters. Always sanitise to snake_case and map back via aliases.

2. Nullable unions require explicit defaults. ["null", "string"] without "default": null is not valid Avro. Always pair them.

3. Empty string and null are not the same thing. CSV has no native null type. Whether an empty cell becomes null or "" in NiFi depends entirely on the Null Value property of the CSVReader. Leave it blank to map empty cells to null — which is what triggers schema rejection for missing required fields.

Part 1 — RTDS Attendance

The Attendance feed has 22 columns. The required core identifies the patient and the submitting organisation. A row missing any of these is clinically meaningless and should be rejected.

Required fields: NHS NUMBER, LOCAL PATIENT IDENTIFIER, NHS NUMBER STATUS INDICATOR CODE, PERSON BIRTH DATE, ORGANISATION IDENTIFIER (CODE OF PROVIDER), PERSON FAMILY NAME, PERSON GIVEN NAME, POSTCODE OF USUAL ADDRESS, PERSON STATED GENDER CODE.

Optional fields: everything else — administrative category, GP details, attendance identifiers, procedure codes, and local version numbers. These are legitimately absent depending on attendance type, pathway, or local system configuration.

{
  "type": "record",
  "name": "RtdsAttendance",
  "namespace": "uk.nhs.rtds.attendance",
  "doc": "RTDS Attendance CSV row schema",
  "fields": [
    { "name": "NHS_NUMBER", "aliases": ["NHS NUMBER"], "type": "string" },
    { "name": "LOCAL_PATIENT_IDENTIFIER", "aliases": ["LOCAL PATIENT IDENTIFIER"], "type": "string" },
    { "name": "NHS_NUMBER_STATUS_INDICATOR_CODE", "aliases": ["NHS NUMBER STATUS INDICATOR CODE"], "type": "string" },
    { "name": "PERSON_BIRTH_DATE", "aliases": ["PERSON BIRTH DATE"], "type": "string" },
    { "name": "ORGANISATION_IDENTIFIER_CODE_OF_PROVIDER", "aliases": ["ORGANISATION IDENTIFIER (CODE OF PROVIDER)"], "type": "string" },
    { "name": "PERSON_FAMILY_NAME", "aliases": ["PERSON FAMILY NAME"], "type": "string" },
    { "name": "PERSON_GIVEN_NAME", "aliases": ["PERSON GIVEN NAME"], "type": "string" },
    { "name": "POSTCODE_OF_USUAL_ADDRESS", "aliases": ["POSTCODE OF USUAL ADDRESS"], "type": "string" },
    { "name": "PERSON_STATED_GENDER_CODE", "aliases": ["PERSON STATED GENDER CODE"], "type": "string" },
    { "name": "ADMINISTRATIVE_CATEGORY_CODE_RADIOTHERAPY", "aliases": ["ADMINISTRATIVE CATEGORY CODE (RADIOTHERAPY)"], "type": ["null", "string"], "default": null },
    { "name": "TRUST_INTERNAL_SYSTEM_PATIENT_ID", "aliases": ["TRUST INTERNAL SYSTEM PATIENT ID"], "type": ["null", "string"], "default": null },
    { "name": "GENERAL_MEDICAL_PRACTITIONER_SPECIFIED", "aliases": ["GENERAL MEDICAL PRACTITIONER (SPECIFIED)"], "type": ["null", "string"], "default": null },
    { "name": "GENERAL_MEDICAL_PRACTICE_CODE_PATIENT_REGISTRATION", "aliases": ["GENERAL MEDICAL PRACTICE CODE (PATIENT REGISTRATION)"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ATTENDANCE_IDENTIFIER", "aliases": ["RADIOTHERAPY ATTENDANCE IDENTIFIER"], "type": ["null", "string"], "default": null },
    { "name": "ADMITTED_PATIENT_ATTENDANCE_INDICATOR", "aliases": ["ADMITTED PATIENT ATTENDANCE INDICATOR"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ATTENDANCE_DATE_AND_TIME", "aliases": ["RADIOTHERAPY ATTENDANCE DATE AND TIME"], "type": ["null", "string"], "default": null },
    { "name": "UNSORTED_PROCEDURE", "aliases": ["UNSORTED PROCEDURE"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ATTENDANCE_PROCEDURE_OPCS", "aliases": ["RADIOTHERAPY ATTENDANCE PROCEDURE (OPCS)"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ATTENDANCE_PROCEDURE_SNOMED_CT", "aliases": ["RADIOTHERAPY ATTENDANCE PROCEDURE (SNOMED CT)"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ADDITIONAL_PROCEDURES", "aliases": ["RADIOTHERAPY ADDITIONAL PROCEDURES"], "type": ["null", "string"], "default": null },
    { "name": "OTHER_ADDITIONAL_RADIOTHERAPY_PROCEDURE", "aliases": ["OTHER ADDITIONAL RADIOTHERAPY PROCEDURE"], "type": ["null", "string"], "default": null },
    { "name": "LOCAL_VERSION_NUMBER_ATTENDANCE", "aliases": ["LOCAL VERSION NUMBER (ATTENDANCE)"], "type": ["null", "string"], "default": null }
  ]
}

Part 2 — RTDS Prescription

The Prescription feed has 46 columns spanning patient identity, clinical diagnosis, treatment planning, dose specification, and beam-level exposure detail. The required core is deliberately narrow — only the fields needed to unambiguously identify the patient, episode, and prescription.

Required fields: NHS NUMBER, LOCAL PATIENT IDENTIFIER, TRUST INTERNAL SYSTEM PATIENT ID, RADIOTHERAPY DIAGNOSIS (ICD), RADIOTHERAPY EPISODE IDENTIFIER, RADIOTHERAPY PRESCRIPTION IDENTIFIER.

Optional fields: everything else. Laterality, plan type, beam energy, SNOMED codes, dose fractions, radioisotope — all depend on treatment modality. External beam, brachytherapy, and radiopharmaceutical treatments have entirely different field populations. Rejecting a row for a missing beam energy on a brachytherapy case would be clinically incorrect.

{
  "type": "record",
  "name": "RtdsPrescription",
  "namespace": "uk.nhs.rtds.prescription",
  "doc": "RTDS Prescription CSV row schema",
  "fields": [
    { "name": "NHS_NUMBER", "aliases": ["NHS NUMBER"], "type": "string" },
    { "name": "LOCAL_PATIENT_IDENTIFIER", "aliases": ["LOCAL PATIENT IDENTIFIER"], "type": "string" },
    { "name": "TRUST_INTERNAL_SYSTEM_PATIENT_ID", "aliases": ["TRUST INTERNAL SYSTEM PATIENT ID"], "type": "string" },
    { "name": "RADIOTHERAPY_DIAGNOSIS_ICD", "aliases": ["RADIOTHERAPY DIAGNOSIS (ICD)"], "type": "string" },
    { "name": "RADIOTHERAPY_EPISODE_IDENTIFIER", "aliases": ["RADIOTHERAPY EPISODE IDENTIFIER"], "type": "string" },
    { "name": "RADIOTHERAPY_PRESCRIPTION_IDENTIFIER", "aliases": ["RADIOTHERAPY PRESCRIPTION IDENTIFIER"], "type": "string" },
    { "name": "RADIOTHERAPY_DIAGNOSIS_SNOMED_CT", "aliases": ["RADIOTHERAPY DIAGNOSIS (SNOMED CT)"], "type": ["null", "string"], "default": null },
    { "name": "TUMOUR_LATERALITY", "aliases": ["TUMOUR LATERALITY"], "type": ["null", "string"], "default": null },
    { "name": "DECISION_TO_TREAT_DATE", "aliases": ["DECISION TO TREAT DATE (RADIOTHERAPY TREATMENT EPISODE)"], "type": ["null", "string"], "default": null },
    { "name": "EARLIEST_CLINICALLY_APPROPRIATE_DATE", "aliases": ["EARLIEST CLINICALLY APPROPRIATE DATE"], "type": ["null", "string"], "default": null },
    { "name": "REFERRAL_DATE", "aliases": ["REFERRAL DATE"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_TREATMENT_REGION", "aliases": ["RADIOTHERAPY TREATMENT REGION"], "type": ["null", "string"], "default": null },
    { "name": "ANATOMICAL_TREATMENT_SITE", "aliases": ["ANATOMICAL TREATMENT SITE (RADIOTHERAPY)"], "type": ["null", "string"], "default": null },
    { "name": "TREATMENT_MODALITY", "aliases": ["TREATMENT MODALITY (RADIOTHERAPY)"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_PRIORITY", "aliases": ["RADIOTHERAPY PRIORITY"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_PRESCRIBED_AUTHORISING_CLINICIAN_CODE", "aliases": ["RADIOTHERAPY PRESCRIBED AUTHORISING CLINICIAN CODE"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_LATERALITY_ANATOMICAL_TREATMENT_SITE", "aliases": ["RADIOTHERAPY LATERALITY (ANATOMICAL TREATMENT SITE)"], "type": ["null", "string"], "default": null },
    { "name": "ROYAL_COLLEGE_OF_RADIOLOGISTS_RCR_CATEGORY", "aliases": ["ROYAL COLLEGE OF RADIOLOGISTS (RCR) CATEGORY"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ROUTES_AND_METHODS_OF_ADMINISTRATION", "aliases": ["RADIOTHERAPY ROUTES AND METHODS OF ADMINISTRATION"], "type": ["null", "string"], "default": null },
    { "name": "PRACTITIONER_LICENCE_HOLDER", "aliases": ["PRACTITIONER LICENCE HOLDER"], "type": ["null", "string"], "default": null },
    { "name": "CODE_OF_ORGANISATION_COMMISSIONED_TO_PROVIDE_ACTIVITY", "aliases": ["CODE OF ORGANISATION COMMISSIONED TO PROVIDE ACTIVITY"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_INTENT_OF_TREATMENT", "aliases": ["RADIOTHERAPY INTENT OF TREATMENT"], "type": ["null", "string"], "default": null },
    { "name": "PRESCRIBED_RADIOTHERAPY_CLINICAL_TRIAL", "aliases": ["PRESCRIBED RADIOTHERAPY CLINICAL TRIAL"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_PLAN_IDENTIFIER", "aliases": ["RADIOTHERAPY PLAN IDENTIFIER"], "type": ["null", "string"], "default": null },
    { "name": "TYPE_OF_PLAN", "aliases": ["TYPE OF PLAN"], "type": ["null", "string"], "default": null },
    { "name": "DATE_OF_PLANNING_APPOINTMENT", "aliases": ["DATE OF PLANNING APPOINTMENT"], "type": ["null", "string"], "default": null },
    { "name": "PLAN_NAME", "aliases": ["PLAN NAME"], "type": ["null", "string"], "default": null },
    { "name": "SPECIALIST_RADIOTHERAPY_TREATMENTS", "aliases": ["SPECIALIST RADIOTHERAPY TREATMENTS"], "type": ["null", "string"], "default": null },
    { "name": "OTHER_SPECIALIST_RADIOTHERAPY_TREATMENTS", "aliases": ["OTHER SPECIALIST RADIOTHERAPY TREATMENTS"], "type": ["null", "string"], "default": null },
    { "name": "UNSORTED_PROCEDURES_PLANNING", "aliases": ["UNSORTED PROCEDURES (PLANNING)"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_PLAN_PROCEDURE_OPCS", "aliases": ["RADIOTHERAPY PLAN PROCEDURE (OPCS)"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_PLAN_PROCEDURE_SNOMED_CT", "aliases": ["RADIOTHERAPY PLAN PROCEDURE (SNOMED CT)"], "type": ["null", "string"], "default": null },
    { "name": "PLAN_ADDITIONAL_PROCEDURES", "aliases": ["PLAN ADDITIONAL PROCEDURES"], "type": ["null", "string"], "default": null },
    { "name": "OTHER_PLAN_ADDITIONAL_PROCEDURES", "aliases": ["OTHER PLAN ADDITIONAL PROCEDURES"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_PRESCRIBED_DOSE", "aliases": ["RADIOTHERAPY PRESCRIBED DOSE"], "type": ["null", "string"], "default": null },
    { "name": "PRESCRIBED_DOSE_UNIT_OF_MEASUREMENT", "aliases": ["PRESCRIBED DOSE UNIT OF MEASUREMENT"], "type": ["null", "string"], "default": null },
    { "name": "PRESCRIBED_FRACTIONS", "aliases": ["PRESCRIBED FRACTIONS"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_ACTUAL_DOSE", "aliases": ["RADIOTHERAPY ACTUAL DOSE"], "type": ["null", "string"], "default": null },
    { "name": "ACTUAL_DOSE_UNIT_OF_MEASUREMENT", "aliases": ["ACTUAL DOSE UNIT OF MEASUREMENT"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_EXPOSURE_IDENTIFIER", "aliases": ["RADIOTHERAPY EXPOSURE IDENTIFIER"], "type": ["null", "string"], "default": null },
    { "name": "MACHINE_IDENTIFIER", "aliases": ["MACHINE IDENTIFIER"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_BEAM_TYPE", "aliases": ["RADIOTHERAPY BEAM TYPE"], "type": ["null", "string"], "default": null },
    { "name": "RADIOTHERAPY_BEAM_ENERGY", "aliases": ["RADIOTHERAPY BEAM ENERGY"], "type": ["null", "string"], "default": null },
    { "name": "BEAM_ENERGY_UNIT_OF_MEASUREMENT", "aliases": ["BEAM ENERGY UNIT OF MEASUREMENT"], "type": ["null", "string"], "default": null },
    { "name": "TIME_AND_DATE_OF_EXPOSURE", "aliases": ["TIME AND DATE OF EXPOSURE"], "type": ["null", "string"], "default": null },
    { "name": "RADIOISOTOPE", "aliases": ["RADIOISOTOPE"], "type": ["null", "string"], "default": null },
    { "name": "RADIOPHARMACEUTICAL_PROCEDURE_SNOMED_CT", "aliases": ["RADIOPHARMACEUTICAL PROCEDURE (SNOMED CT)"], "type": ["null", "string"], "default": null }
  ]
}

Configuring the CSVReader

The schema alone is not enough. These are the critical settings on the CSVReader controller service:

The Null Value property is the linchpin. When left blank, NiFi converts any empty CSV cell — whether written as "" or just , — to null in the parsed record. That null will then fail validation against any field typed as plain "string", routing the record to the invalid relationship automatically.

How validation flows in the pipeline

With both CSVReaders configured, the pipeline shape is the same for both feeds:

[GetFile]
    │
    ▼
[SplitRecord]  ◄── CSVReader (schema above) + JSONRecordSetWriter
    │
    ├── valid   ──► downstream processing
    │
    └── invalid ──► dead-letter queue / alert

Every row missing a required demographic or identity field exits through invalid before touching anything downstream. Every row with a missing SNOMED CT code, beam energy, or plan name passes through valid — because those fields are legitimately nullable.

A note on alias support: alias-based header matching is reliable when the CSV header is present and consistent, which is guaranteed for RTDS exports. If you encounter a NiFi version where aliases are not honoured, the safe fallback is positional mapping — NiFi maps the first schema field to column 1, the second to column 2, and so on. Since neither RTDS header ever changes, positional mapping works equally well.

Summary

Validating RTDS data in NiFi 2 requires aligning three things: a correctly structured Avro schema with legal field names and proper null union defaults, CSVReader settings that translate empty cells to null, and a clear model of which fields are clinically required versus legitimately optional. Once those three are in place, NiFi does the rest — routing invalid records away from your pipeline at the point of ingestion, every time, for both feeds.

Built and tested on Apache NiFi 2 · NHS RTDS Attendance & Prescription datasets · Avro 1.11