🚀 Streaming & Event Pipeline

Zynka ONE is designed for real‑time data ingestion. Every OCPI call (locations, sessions, CDRs, commands) is wrapped in a common envelope and published to Azure Event Hubs. Downstream consumers—including analytics dashboards, billing services and alerting rules—subscribe to this stream via Microsoft Fabric Real‑Time Intelligence (RTI).

Envelope schema

All events sent by Zynka ONE follow a uniform schema. This ensures that different modules (e.g. Sessions, CDRs, Commands) can be consumed consistently and that multi‑tenant data can be partitioned efficiently. The envelope fields include:

Field	Description
`ts_utc`	UTC timestamp when the event was created.
`tenantId`	Identifier of the tenant (the CPO or hub owning the data). Used as the first part of the partition key.
`module`	Type of OCPI object (`locations`, `tariffs`, `sessions`, `cdrs`, `evse_status`, `commands`, etc.).
`entityId`	ID of the underlying object (e.g. Location ID, Session ID, EVSE UID). Combined with `tenantId` to form the partition key.
`x_request_id`	Unique request ID from the incoming HTTP header for traceability.
`x_correlation_id`	Correlation ID across microservices. Useful when a single user action triggers multiple events.
`schema_version`	Version of the envelope schema (e.g. `1.0`).
`payload`	The OCPI object as received or normalised. Fields follow the OCPI spec for that module.
`hash`	SHA‑256 hash of the payload for idempotency checks and tamper detection.

Events are published using the partition key ${tenantId}|${entityId}. This guarantees ordering per entity and spreads load across Event Hub partitions.

Example

{
  "ts_utc": "2025-01-14T10:23:45Z",
  "tenantId": "INZYN",
  "module": "sessions",
  "entityId": "IN*ZYN*E000001",
  "x_request_id": "987e4567-e89b-12d3-a456-426614174000",
  "x_correlation_id": "123e4567-e89b-12d3-a456-426614174111",
  "schema_version": "1.0",
  "payload": { /* OCPI Session object */ },
  "hash": "sha256:abcd..."
}

Event Hubs setup

Zynka ONE deploys an Event Hubs namespace with one hub dedicated to OCPI events. Key guidelines:

Partitions: Use 8–16 partitions to handle high throughput. Partition keys (tenant | entity) ensure order per connector while distributing load.
Consumer groups: Create separate consumer groups for analytics, billing, monitoring and third‑party integrations. Consumer groups allow independent reading offsets without interfering with one another.
Managed Identity: The AKS pods publishing events are assigned a managed identity with Azure Event Hubs Data Sender role. Downstream services use Data Receiver role.
Retry and store‑and‑forward: When publishing fails (e.g. network outage), events are stored locally and retried until delivery succeeds. This ensures that no session or CDR is lost. Store‑and‑forward is also recommended when pulling sessions or CDRs from hubs【826352846409834†L1369-L1378】.

Push vs Pull

OCPI supports both Push and Pull mechanisms. For production systems, the spec recommends using Push for real‑time updates【343458627312853†L1861-L1905】. Zynka ONE publishes all events to Event Hubs (Push). However, certain partners may still pull sessions or CDRs periodically. If implementing Pull:

Pagination: Support offset and limit parameters. Use them to read events in chronological order.
Store‑and‑forward: Maintain a queue of events to deliver if the receiving partner is temporarily unreachable.【826352846409834†L1369-L1378】.
Idempotency: Use the id and last_updated fields of OCPI objects to prevent duplicate processing.

Stream processing with Fabric RTI

After events are ingested, Zynka ONE uses Microsoft Fabric Real‑Time Intelligence (RTI) to process and store the data:

Eventstream: Consumes Event Hub messages and performs lightweight transformations (e.g. flattening nested JSON, adding ingestion timestamps). Eventstream routes events to downstream services or databases.
Eventhouse: Stores events in a Delta Lakehouse. Data is partitioned by date and module for efficient queries. Eventhouse supports schema evolution and time travel.
Materialised views: Using Kusto Query Language (KQL), Zynka ONE creates materialised views to compute aggregates such as the latest EVSE status or total energy per charger. Example:

.create-or-alter materialized-view current_session_progress on table ocpi_events_raw {
  ocpi_events_raw
  | where module == "sessions"
  | summarize arg_max(ts_utc, payload.kwh, payload.total_cost) by entityId
  | project session_id = entityId, kwh = payload_kwh, total_cost
}

Tips for integrators

Traceability: Always populate X-Request-ID and X-Correlation-ID headers. These values propagate into the envelope and allow end‑to‑end traceability across microservices【826352846409834†L1224-L1240】.
Partition awareness: If you host your own consumers, use the same partition key strategy to guarantee order. Avoid consuming from multiple partitions in a single transaction when order matters.
Error handling: Implement retry with exponential backoff. If an event cannot be processed, move it to a dead‑letter table for later inspection.
Security: Encrypt Event Hub connections with TLS; rotate keys regularly. Use Azure Monitor to track ingress/egress metrics and set alert thresholds.

The streaming pipeline ensures that every OCPI interaction—whether a token authorization, a session update, a tariff change or a UPI payment—is recorded and available for analytics and auditing within seconds.

Envelope schema​

Example​

Event Hubs setup​

Push vs Pull​

Stream processing with Fabric RTI​

Tips for integrators​