Daten verarbeiten
Prüfen und vereinheitlichen Sie die geladenen Rohdaten. Wenn Quell- und Zielstruktur voneinander abweichen, legen Sie das Mapping hier fest und erzeugen einen validierten Datensatz für den anschließenden Speicherweg.
Diese Anleitung beginnt mit Rohdaten in msg.payload. Wie HTTP-, MQTT-, S3- oder Datenbankdaten dorthin gelangen, zeigt Daten laden. Gespeichert wird auf der nächsten Seite; hier endet der Flow-Abschnitt mit fachlich geprüften Daten und einem nachvollziehbaren Nachrichtenvertrag.
Rohdaten prüfen
Array, Pflichtfelder und Datentypen technisch kontrollieren.
Mappingbedarf klären
Quellstruktur und benötigte fachliche Struktur vergleichen.
Daten zuordnen
Felder, Zeiten, Einheiten und Kennungen sichtbar abbilden.
Datensatz validieren
Einen eindeutigen Vertrag für den Speicherweg erzeugen.
Mit einem bekannten Eingang und Ziel beginnen
Vor der Verarbeitung müssen der geladene Beispieldatensatz und der vorgesehene Zielweg bekannt sein. Die technische Zielkonfiguration folgt erst unter Daten speichern, aber ihre fachlichen Anforderungen bestimmen bereits die Verarbeitung.
Geladene Rohdaten
Ein repräsentativer und ein bewusst fehlerhafter Datensatz liegen in msg.payload vor.
Vorgesehener Speicherweg
Stellio, FROST, PostgreSQL oder S3 sowie die erwartete fachliche Struktur sind bekannt.
Erfolgs- und Fehlerfälle
Pflichtwerte, zulässige Bereiche und Verhalten bei ungültigen Datensätzen sind geklärt.
Stabile Identität
Schlüssel, fachlicher Zeitpunkt und Verhalten bei erneuter Verarbeitung sind festgelegt.
Eine unveränderte Datei oder eine bereits passende Tabellenstruktur kann ohne eigenes Mapping weitergegeben werden. Sobald Felder umbenannt, Typen oder Einheiten umgerechnet, Kennungen gebildet oder Daten in ein fachliches Modell wie NGSI-LD beziehungsweise SensorThings überführt werden, gehört das Mapping in diese Aufgabe.
Rohdaten technisch und fachlich prüfen
Die Ladewege der vorherigen Seite liefern ein Array in msg.payload. Prüfen Sie zuerst die technische Verwendbarkeit und danach die fachlichen Regeln. Ein erfolgreicher Parser allein sagt noch nicht, dass ein Datensatz verwendet werden darf.
| Prüfung | Beispiel | Reaktion bei Abweichung |
|---|---|---|
| technische Struktur | msg.payload ist ein nicht leeres Array | technischer Fehlerpfad |
| Pflichtwerte | Messstelle, Zeitpunkt und Zählwert sind vorhanden | Datensatz zurückweisen |
| Datentyp | Zählwert ist eine nicht negative Ganzzahl | fachlicher Fehlerpfad |
| Zeit | Zeitpunkt ist parsebar und wird in UTC normalisiert | Datensatz zurückweisen |
| Wiederholung | gleiche Messstelle und gleicher Zeitpunkt bilden dieselbe Identität | idempotent weiterführen |
Führen Sie technische und fachliche Fehler getrennt. Protokollieren Sie Lauf-ID, Quelle und eine ungefährliche Fehlerbeschreibung, aber keine Passwörter, Tokens oder vollständigen schützenswerten Payloads.
Mapping bei Bedarf festlegen
Vergleichen Sie die realen Quellfelder mit dem vereinbarten Nachrichtenvertrag für die weitere Verarbeitung. Die vier Ladebeispiele verwenden bereits dieselbe Struktur; für sie ist keine Umbenennung erforderlich:
| Bedeutung | Vereinbarter Rohdatenvertrag | Zusätzliche Verarbeitung |
|---|---|---|
| Messstelle | counting_station_id | als nicht leere Zeichenkette prüfen |
| Messzeitpunkt | observed_at | parsen und als ISO-8601 in UTC normalisieren |
| Fahrzeuganzahl | vehicle_count | in eine nicht negative Ganzzahl umwandeln |
Ein Mapping wird erst benötigt, wenn eine reale Quelle davon abweicht. Liefert eine API beispielsweise countingStationId, observedAt und vehicleCount, halten Sie die Zuordnung zu counting_station_id, observed_at und vehicle_count ausdrücklich fest, bevor der Flow sie umbenennt.
Dokumentieren Sie für jedes abweichende Feld Quellpfad, Zielattribut, Umwandlung und Verhalten bei fehlendem oder ungültigem Wert. Ergänzen Sie außerdem die Entscheidungen, die mehrere Felder betreffen:
- Wie wird eine stabile fachliche Kennung gebildet?
- Welcher Zeitpunkt beschreibt die Beobachtung und welche Zeitzone gilt?
- Welche Einheit und welcher zulässige Wertebereich werden erwartet?
- Was geschieht mit fehlenden, verspäteten oder doppelt zugestellten Daten?
- Führt eine Wiederholung zu einer Aktualisierung oder zu einem neuen Datensatz?
Ein ähnlich benanntes Feld darf nicht still für einen anderen Sachverhalt verwendet werden. Halten Sie die Zuordnung neben dem Flow fest und versionieren Sie sie gemeinsam mit ungefährlichen Beispieldaten. So kann eine weitere Person die Function Node gegen dieselben Regeln prüfen.
Verarbeitung in Node-RED umsetzen
Teilen Sie den Flow in erkennbare Abschnitte. Der eigentliche Writer gehört noch nicht hierher:
geladene Rohdaten → technische Prüfung → fachliche Prüfung und Mapping → validierter Ausgang
↘ kontrollierter Fehlerpfad
Prüfen Sie vor eigener Standardlogik, ob die HYPERTEGRITY Node-RED-Bibliothek einen benötigten Baustein bereits bereitstellt. Gemeinsame Nodes eignen sich für wiederkehrende technische Aufgaben; projektspezifische fachliche Regeln bleiben sichtbar im Flow.
Das folgende Beispiel validiert den gemeinsamen Rohdatenvertrag. Es akzeptiert den gültigen Zahlenwert 0, weist leere Werte zurück, normalisiert den Zeitpunkt und erhält den Quellenbezug:
const rows = Array.isArray(msg.payload) ? msg.payload : [msg.payload];
if (rows.length === 0) {
throw new Error("Der Ladeweg hat keine Rohdatensätze geliefert");
}
msg.payload = rows.map((row, index) => {
const countingStationId = row.counting_station_id;
const observedAt = row.observed_at;
const rawVehicleCount = row.vehicle_count;
if (
!countingStationId
|| !observedAt
|| rawVehicleCount === null
|| rawVehicleCount === undefined
|| (typeof rawVehicleCount === "string" && rawVehicleCount.trim() === "")
) {
throw new Error(`Verkehrsdatensatz ${index + 1}: Pflichtwert fehlt`);
}
const observedAtMillis = Date.parse(observedAt);
const vehicleCount = Number(rawVehicleCount);
if (Number.isNaN(observedAtMillis)) {
throw new Error(`Verkehrsdatensatz ${index + 1}: Zeitpunkt ist ungültig`);
}
if (!Number.isInteger(vehicleCount) || vehicleCount < 0) {
throw new Error(`Verkehrsdatensatz ${index + 1}: Zählwert ist ungültig`);
}
return {
counting_station_id: String(countingStationId),
observed_at: new Date(observedAtMillis).toISOString(),
vehicle_count: vehicleCount
};
});
msg.context = {
runId: `${new Date().toISOString()}-${msg.payload[0].counting_station_id}`,
source: msg.source ?? {type: "unknown"}
};
return msg;
Der Ausgang ist ein kleiner Nachrichtenvertrag zwischen Verarbeitung und Speicherung:
{
"payload": [
{
"counting_station_id": "zaehlstelle-017",
"observed_at": "2026-08-19T08:00:00Z",
"vehicle_count": 126
}
],
"context": {
"runId": "2026-08-20T08:03:12.000Z-zaehlstelle-017",
"source": {
"type": "s3",
"bucket": "verkehr-rohdaten",
"objectKey": "verkehrszaehlung-2026-08-19.csv"
}
}
}
Dieser Vertrag ist kein zusätzliches Plattformformat. Er hält nur fest, was der folgende Speicherabschnitt erwarten darf. Die zielabhängige Übersetzung in eine NGSI-LD-Entität, SensorThings-Observation, SQL-Anweisung oder Datei erfolgt auf der nächsten Seite, ohne die fachliche Bedeutung erneut zu verändern.
Verarbeitung prüfen und übergeben
Prüfen Sie die Verarbeitung mit gespeicherten Beispieldaten und deaktiviertem Produktiveingang:
| Testfall | Erwarteter Nachweis |
|---|---|
| vollständiger Datensatz | validierter Datensatz mit normalisiertem Zeitpunkt |
| fehlendes Pflichtfeld | kontrollierter fachlicher Fehler; kein Ausgang zum Writer |
| falscher Datentyp oder negativer Wert | nachvollziehbare Ablehnung |
| erneute identische Eingabe | identischer fachlicher Schlüssel und Zeitpunkt |
| Logging | Lauf-ID und Ursache, aber keine Secrets oder vollständigen geschützten Daten |
Exportieren und versionieren Sie Flow, benötigte Paketversionen, Mappingregeln und ungefährliche Beispieldaten. Die Verarbeitung ist abgeschlossen, wenn der validierte Vertrag für alle Erfolgs- und Fehlerfälle reproduzierbar ist.