Daten speichern
Übergeben Sie den validierten Datensatz an das vereinbarte Ziel. Die Beispiele zeigen vier häufige Speicherwege über Stellio, FROST, PostgreSQL und einen S3-kompatiblen Objektspeicher.
Diese Anleitung beginnt mit dem unter Daten verarbeiten geprüften Nachrichtenvertrag. Fachliche Bedeutung, Pflichtwerte und Kennungen sind bereits geklärt. Hier wird daraus das Format des Zielsystems gebildet, der Schreibzugriff ausgeführt und der tatsächliche Erfolg nachgewiesen.
{
"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"
}
}
Ein Flow wählt sein Ziel nicht während der Ausführung beliebig aus. Zielkomponente, Datenraum, Datastream, Tabelle oder Bucket sowie die erlaubte Schreiboperation müssen vorher feststehen. Verwenden Sie einen zweckgebundenen technischen Zugang und keine persönlichen Passwörter oder dauerhaft hinterlegten Benutzertokens.
Vier beispielhafte Zielwege
Wählen Sie den Tab für das bereits vereinbarte Ziel. Die Beispiele sind keine freie Speicherentscheidung: Sie zeigen, wie derselbe validierte Datensatz für unterschiedliche, fachlich begründete Zielwege geschrieben wird.
- Stellio / NGSI-LD
- FROST / SensorThings
- PostgreSQL
- Datei / S3
Passt, wenn der aktuelle Zustand einer fachlichen Entität über NGSI-LD bereitgestellt werden soll. Entitätstyp, Attribute, stabile ID und JSON-LD-Kontext müssen zuvor in der Verarbeitung festgelegt sein.
1. NGSI-LD-Entitäten bilden
Das Beispiel verwendet einen projektspezifisch abgestimmten Typ. Es zeigt die technische Übersetzung des validierten Vertrags und ersetzt keine Modellentscheidung:
const contextUrl = "https://context.<DOMAIN>/contexts/traffic-count.jsonld";
msg.payload = msg.payload.map((row) => ({
id: `urn:ngsi-ld:TrafficCount:${row.counting_station_id}`,
type: "TrafficCount",
dateObserved: {
type: "Property",
value: row.observed_at
},
vehicleCount: {
type: "Property",
value: row.vehicle_count
},
"@context": [contextUrl]
}));
return msg;
2. Entitäten an Stellio übergeben
Mit der freigegebenen HYPERTEGRITY-Bibliothek sollte die Struktur bevorzugt über ngsi-builder aufgebaut und mit post-to-stellio geschrieben werden:
validierter Datensatz → ngsi-builder → post-to-stellio → switch(statusCode)
↘ kontrollierter Fehlerpfad
Der veröffentlichte post-to-stellio-Node der Bibliotheksversion 0.1.2 setzt den
Datenraum-Header, unterstützt aber selbst keinen OAuth-Token für einen geschützten APISIX-Pfad. Verwenden Sie
ihn nur mit einem dafür vorgesehenen internen Stellio-Zugang. Für einen geschützten externen Pfad benötigen
Sie einen technischen Client und einen geeigneten HTTP-/Authentifizierungsweg.
Prüfen Sie nach dem Writer msg.statusCode. Auch 4xx- und 5xx-Antworten müssen in den Fehlerpfad laufen. Der Upsert muss bei derselben stabilen Entity-ID denselben fachlichen Zustand aktualisieren, statt eine zweite Entität anzulegen.
der Writer den erwarteten Erfolgsstatus liefert und die Entität anschließend mit ihrer vollständigen ID im vorgesehenen Datenraum gelesen werden kann. Ein erfolgreicher Node-Durchlauf ohne API-Nachweis genügt nicht.
Passende Referenzen
Passt, wenn einzelne Messwerte zeitlich geordnet einem bereits eingerichteten SensorThings-Datastream zugeordnet werden. Thing, Sensor, ObservedProperty und Datastream müssen vor der ersten Observation vorbereitet sein.
1. Pro Datensatz eine Observation bilden
const datastreamByStation = {
"zaehlstelle-017": 42,
"zaehlstelle-021": 43
};
const messages = msg.payload.map((row) => {
const datastreamId = datastreamByStation[row.counting_station_id];
if (!datastreamId) {
throw new Error(`Kein Datastream für ${row.counting_station_id} zugeordnet`);
}
return {
method: "POST",
url: "https://api.<DOMAIN>/frost-http/v1.1/Observations",
headers: {"Content-Type": "application/json"},
payload: {
phenomenonTime: row.observed_at,
result: row.vehicle_count,
Datastream: {"@iot.id": datastreamId}
}
};
});
return [messages];
Die Function sendet jede Observation als eigene Nachricht an ihrem ersten Ausgang.
2. Observation schreiben
Stellen Sie http request auf set by msg.method und lassen Sie das URL-Feld leer, damit Methode und URL aus der Nachricht verwendet werden. Der eingerichtete Authentifizierungsweg ergänzt ein kurzlebiges Access Token mit api:write; das Token darf nicht in Function Nodes, Debug-Ausgaben oder Flow-Exporten erscheinen.
Ein POST /Observations legt bei jedem erfolgreichen Aufruf eine neue Observation an. Leiten Sie deshalb aus Quellenkennung und fachlichem Zeitpunkt einen Verarbeitungsnachweis ab und unterdrücken Sie bereits verarbeitete Nachrichten. Wiederholen Sie einen unklar abgebrochenen POST nicht blind.
FROST 201 Created liefert und die Observation anschließend im erwarteten Datastream mit
fachlichem Zeitpunkt und Ergebnis gelesen werden kann. Behandeln Sie 401, 403,
404 und 5xx getrennt.
Passende Referenzen
Passt, wenn die validierten Daten in einer vorbereiteten relationalen Tabelle gespeichert werden sollen. Zieltabelle, Spaltentypen und ein eindeutiger Schlüssel müssen vor dem Flow feststehen.
1. Parametrisierten Upsert vorbereiten
Das Beispiel aktualisiert denselben fachlichen Datensatz bei erneuter Verarbeitung:
const query = `
INSERT INTO public.traffic_counts_curated (
counting_station_id,
observed_at,
vehicle_count
)
VALUES ($1, $2::timestamptz, $3)
ON CONFLICT (counting_station_id, observed_at)
DO UPDATE SET vehicle_count = EXCLUDED.vehicle_count;
`;
const messages = msg.payload.map((row) => ({
query,
params: [row.counting_station_id, row.observed_at, row.vehicle_count],
context: msg.context
}));
return [messages];
2. Zeilen schreiben
Verbinden Sie die Function mit der freigegebenen postgresql-Node. Sie benötigt eine administrativ bereitgestellte Verbindung und eine technische Rolle mit Schreibrecht ausschließlich auf das vorgesehene Schema beziehungsweise die Zieltabelle. Tabellen- und Spaltennamen dürfen nicht aus unkontrollierten Nachrichten zusammengesetzt werden.
Ist postgresql nicht in der Palette vorhanden, lassen Sie die freigegebene Node durch die Administration installieren. Das Vorgehen steht unter Node-RED-Erweiterungen bereitstellen.
der Writer genau die erwartete Anzahl Zeilen einfügt oder aktualisiert und derselbe Testdatensatz bei einer Wiederholung keinen unerwünschten Duplikateintrag erzeugt. Prüfen Sie das Ergebnis anschließend lesend über die vorgesehene Tabelle oder View.
Passende Referenzen
Passt, wenn die validierten Daten als Export, weiterzugebende Datei oder vereinbartes Objekt abgelegt werden sollen. Bucket, Präfix, Format und Benennungsregel müssen feststehen.
1. Export und Object Key bilden
const runDate = "2026-08-20";
msg.filename = `exports/${runDate}/traffic-counts.json`;
msg.payload = JSON.stringify(msg.payload, null, 2);
return msg;
2. Objekt schreiben
Konfigurieren Sie in amazon s3 out den Bucket verkehr-verarbeitet statisch. Der Beispiel-Node prüft den Bucket beim Start und benötigt deshalb neben PutObject zusätzlich s3:ListBucket; msg.bucket allein genügt bei diesem Paket nicht. Kann dieses zusätzliche Recht nicht vertretbar begrenzt werden, muss die Administration einen anderen freigegebenen S3-Ausgabeknoten wählen.
Der technische Zugang erhält Schreibrecht nur auf den Ziel-Bucket beziehungsweise das vereinbarte Präfix. Für einen externen S3-Speicher ändern sich Verbindung, Endpoint und Credentials, nicht die Bildung des Exports. Ein externer Bucket ist nicht in HYPERSENSUS sichtbar.
Ist amazon s3 out nicht in der Palette vorhanden, lassen Sie die freigegebene Node durch die Administration installieren. Das Vorgehen steht unter Node-RED-Erweiterungen bereitstellen.
das Objekt unter dem erwarteten Bucket und vollständigen Object Key vorhanden ist, Größe und Format stimmen und eine Wiederholung nicht unbeabsichtigt einen fachlich anderen Lieferstand überschreibt. Prüfen Sie ein UDSP-MinIO-Ziel über HYPERSENSUS oder MinIO, einen externen Bucket über dessen S3-Zugang.
Passende Referenzen
Speicherung abschließen
Führen Sie die erste Speicherung mit gespeicherten Testdaten und deaktiviertem Produktiveingang durch. Prüfen Sie Erfolgsfall, ungültige Daten, wiederholte Zustellung und ein nicht erreichbares Ziel. Schalten Sie die laufende Quelle erst zu, wenn Writer und Fehlerpfad nachweislich funktionieren.
Versionieren Sie Flow, benötigte Node-Pakete und ungefährliche Testdaten. Dokumentieren Sie Ziel, technische Identität, erlaubte Schreiboperation, Wiederanlaufverhalten und den fachlichen Nachweis am Ziel.