System Design: Near Real-Time EV Charging (OCPP) Pipeline on Databricks

A model system-design answer: near real-time EV charging events on Databricks with dedup, late data, quarantine and a sessions Gold table.

Same series as Parts 1-10. This is a model answer for the scenario: OCPP JSON events from charge points (StartTransaction, MeterValues, StopTransaction, StatusNotification, Heartbeat), with duplicates, business duplicates, out-of-order and late events, schema evolution, corrupted records and text/data-quality errors, ending in analytical tables including a charging_sessions Gold table.

It is structured the way to deliver it live: a 3-minute spoken answer first, then the deep dive you can expand on demand.

Caveat: items marked [verify] are Databricks features that were renamed or changed recently (naming of Lakeflow/DLT, some streaming APIs, VARIANT). Check the current docs before stating them as fact.

0. HOW TO RUN THE 30-MINUTE INTERVIEW

Rounds like this are usually about architecture decisions, table design, edge cases and reasoning, not coding. Spend your time on why, and use code only as short illustrations.

1. CLARIFYING QUESTIONS AND ASSUMPTIONS

Ask these first (interviewers score this), then state your assumptions explicitly so the design is anchored.

Questions

1. What does "near real-time" mean here: seconds, 1-5 minutes, or 15 minutes? Who consumes it (operations dashboards, billing, analysts)?
2. Volume: how many charge points, events per second, peak factor?
3. OCPP version (1.6J vs 2.0.1)? Who normalizes the messages into the flat JSON events we see?
4. How late can events arrive? (Charge points buffer messages while offline and replay them later, so hours or days are realistic.)
5. What counts as a business duplicate, and which copy wins?
6. Is energy used for billing (needs final, auditable numbers) or only for operational insight?
7. Retention, PII (the user token/idTag is a personal identifier), and regulatory constraints?

Assumptions I will state

• Freshness target: Gold sessions within \~5 minutes of event arrival; daily statistics refreshed within \~15 minutes.
• Scale: 100k charge points; about 20% charging at peak, MeterValues every 60 s gives roughly 330 events/s; Heartbeats every 5 min add another \~330 events/s. Call it \~700 events/s average and plan for 10x peaks. This is a small-to-medium streaming load, so favor simplicity and correctness over exotic tuning.
• Events arrive via a broker (Azure Event Hubs with Kafka endpoint, or Kafka), partitioned by charge_point_id.
• A gateway (OCPP central system) already converts OCPP frames to flat JSON with the fields in the sample. Subtle point to mention: in OCPP 1.6 the StartTransaction request does not contain a transaction ID; the central system assigns it in the response, so the gateway must enrich events with it. In 2.0.1 the charge point supplies it and also sends a per-transaction sequence number (seqNo), which helps ordering.
• Late data can arrive up to 7 days after the event; sessions with no events for 24 hours are considered abandoned.

2. THE 3-MINUTE SPOKEN ANSWER

"I'd build a medallion pipeline on Delta Lake, with streaming for Bronze and Silver and a recompute-by-key pattern for Gold.

Ingestion: charge-point events go to Event Hubs, partitioned by charge point ID so per-device order is preserved within a partition. Databricks Structured Streaming, running as a Lakeflow Declarative Pipeline or a plain Spark job, reads the topic and writes Bronze: an append-only Delta table holding the raw JSON payload as a string plus Kafka metadata and ingestion time. Nothing is parsed, dropped or fixed in Bronze, so I can always replay.

Silver is where correctness lives. I parse with an explicit schema, normalize text and units, and validate mandatory fields and ranges. Failing records go to a quarantine table with reason codes and the original payload, and a repair-and-replay path. Valid records are deduplicated twice: technical duplicates by event_id (retries), and business duplicates by a business key hash over the event's natural identity. Duplicates are logged to an audit table, not silently dropped. Silver is event-time, append-only and never discards late data. Watermarks only bound streaming state; they never decide what data is acceptable.

Gold sessions: because the meter readings are cumulative, a session's energy is a pure function of all its events. So instead of fragile stateful logic, for each micro-batch I find the affected sessions, recompute them from Silver, and MERGE them into the sessions table. That's naturally idempotent, and it makes out-of-order and late events a non-issue: a late MeterValues just triggers a recompute. Energy is meter_stop - meter_start once the StopTransaction exists. Until then it's provisional, from the last meter value, with energy_is_final = false and a status of IN_PROGRESS. A sweeper marks sessions with no activity for 24 hours as timed out.

Other Gold tables: daily charging statistics (final and provisional energy kept separate), a connector status history/current-status table, and a charge-point health table fed by heartbeats.

Operational side: schema evolution is handled by keeping raw payloads, parsing with versioned schemas and a rescued-data column, and changing curated schemas only through reviewed code. Every layer is replay-safe: Bronze by offsets, Silver by the two dedup keys, Gold by deterministic recompute. A nightly reconciliation job recomputes the last N days to heal anything missed, and monitoring covers freshness, quarantine and duplicate rates, late-arrival lag, and a row-accounting check that every Bronze row ended up in exactly one place."

3. ARCHITECTURE

 Charge points (OCPP 1.6J / 2.0.1 over WebSocket)
        │
        ▼
 OCPP gateway (normalizes frames to flat JSON, adds charge_point_id from connection, txn id)
        │
        ▼
 Azure Event Hubs (Kafka endpoint)  ──► Event Hubs Capture ──► ADLS (long-term raw archive, replay beyond retention)
   key = charge_point_id (per-device ordering within a partition)
        │
        ▼
 ┌────────────────────── Databricks (Unity Catalog, Delta) ───────────────────────┐
 │                                                                                │
 │ BRONZE  raw_ocpp_events      append-only; raw JSON string + Kafka metadata     │
 │    │                                                                           │
 │    ▼   parse · normalize · validate                                            │
 │ SILVER                                                                         │
 │    ├─ charging_events         valid, typed, deduplicated, event-time           │
 │    ├─ quarantine_events       invalid records + reason codes + payload         │
 │    ├─ duplicate_events_audit  technical and business duplicates removed        │
 │    └─ (CDF enabled on charging_events)                                         │
 │    │                                                                           │
 │    ▼   recompute affected keys · MERGE                                         │
 │ GOLD                                                                           │
 │    ├─ charging_sessions            1 row per session (final or provisional)    │
 │    ├─ charging_stats_daily         by charge point / day                       │
 │    ├─ connector_status_history     intervals (valid_from, valid_to)            │
 │    ├─ connector_current_status     latest state per connector                  │
 │    └─ charge_point_health          last heartbeat / liveness                   │
 │                                                                                │
 │ OPS   dq_metrics · reconciliation_log · pipeline run log                       │
 └────────────────────────────────────────────────────────────────────────────────┘

Why a broker between the gateway and Databricks: it decouples producers from consumers, absorbs bursts, gives replay within its retention, and preserves per-key order. Why also archive via Capture: broker retention is days; the archive lets us rebuild Bronze after a disaster or a very late bug discovery.

Compute choice: Lakeflow Declarative Pipelines (the evolution of Delta Live Tables) [verify naming] for Bronze→Silver gives managed checkpoints, built-in expectations (data-quality rules), lineage and simpler operations. Plain Structured Streaming with foreachBatch is the alternative when I need custom logic (the session recompute). In practice: declarative pipeline for Bronze/Silver, a foreachBatch job (or a materialized view) for the session logic.

4. BRONZE LAYER: raw_ocpp_events

Purpose: a faithful, immutable, replayable record of everything received.

CREATE TABLE bronze.raw_ocpp_events (
  raw_payload        STRING    COMMENT 'original JSON exactly as received',
  kafka_topic        STRING,
  kafka_partition    INT,
  kafka_offset       BIGINT,
  kafka_key          STRING    COMMENT 'charge_point_id if the gateway set it',
  broker_ts          TIMESTAMP COMMENT 'enqueue time at the broker',
  _ingest_ts         TIMESTAMP COMMENT 'when Databricks wrote the row',
  _ingest_date       DATE      GENERATED ALWAYS AS (CAST(_ingest_ts AS DATE))
)
USING DELTA
CLUSTER BY (_ingest_date)
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.logRetentionDuration'       = 'interval 30 days'
);

Design decisions and reasoning

• Payload as a string (or VARIANT), not parsed columns: Bronze is immune to schema changes and corrupted JSON. A malformed record still lands, so nothing is lost before validation. VARIANT is an option for semi-structured storage [verify availability in your runtime]; a string is the most portable.
• Kafka coordinates (topic, partition, offset) form the natural unique identity of a Bronze row. They make Bronze reprocessing traceable and let quarantine and audit rows point back to the exact source record.
• Append-only, no dedup, no cleaning. Duplicates in Bronze are fine; it is the log of what arrived.
• Cluster/partition by ingest date, not event date: it matches how data arrives and keeps late events from scattering writes across old partitions.
• Retention: keep Bronze long enough to replay (for example 90-400 days depending on regulation), with Heartbeat rows eligible for a shorter retention since they carry little analytical value; the Capture archive holds the long tail.
• Delivery semantics: Structured Streaming plus checkpoints gives at-least-once into Bronze. A replay after a crash can write a batch twice; that's acceptable because downstream layers deduplicate. If strict Bronze uniqueness is wanted, writing with Delta's idempotent-write options (txnAppId/txnVersion) in foreachBatch makes each batch write exactly-once.

bronze = (spark.readStream.format("kafka")
    .options(**kafka_opts)                       # Event Hubs Kafka endpoint
    .option("subscribe", "ocpp-events")
    .option("startingOffsets", "earliest")
    .option("maxOffsetsPerTrigger", 500_000)     # bound batch size, protects against backlog spikes
    .load()
    .select(
        col("value").cast("string").alias("raw_payload"),
        col("topic").alias("kafka_topic"), col("partition").alias("kafka_partition"),
        col("offset").alias("kafka_offset"), col("key").cast("string").alias("kafka_key"),
        col("timestamp").alias("broker_ts"),
        current_timestamp().alias("_ingest_ts")))

(bronze.writeStream.format("delta")
    .option("checkpointLocation", ckpt_bronze)
    .trigger(processingTime="30 seconds")
    .toTable("bronze.raw_ocpp_events"))

5. SILVER LAYER

Silver turns raw into trustworthy, typed, deduplicated events. Three outputs: charging_events (good), quarantine_events (bad), duplicate_events_audit (removed duplicates).

5.1 Parse and normalize

• Parse with an explicit schema per event type and schema_version; keep a _rescued_data (or extra_fields) column for fields the schema doesn't know. Unknown-field rate is a monitored metric (Section 9).
• Normalize deterministically:
  • Trim and case-normalize text; map status strings to the enum ("charging", "Charging " → Charging).
  • Parse timestamps to UTC with a defined rule; reject unparsable ones.
  • Normalize units: OCPP meter values may be Wh or kWh; store a canonical meter_wh.
  • Normalize charge_point_id/connector_id formats.
• Recover before rejecting. If charge_point_id is NULL in the payload, try the Kafka key, which the gateway may have set from the WebSocket URL. Only quarantine when it cannot be recovered. This is an example of a safe, deterministic fix; guessing is not.

Rule of thumb for text/data-quality errors: fix what is deterministic and safe, quarantine what is ambiguous, and never silently guess.

5.2 Validation rules (declarative)

In a declarative pipeline these are expectations. The robust pattern is to compute an is_valid flag and a violations array, then route: valid rows continue, invalid rows go to the quarantine table (instead of using drop-only expectations that discard evidence).

5.3 Quarantine table

CREATE TABLE silver.quarantine_events (
  quarantine_id   STRING,                -- hash(topic, partition, offset)
  raw_payload     STRING,
  kafka_topic     STRING, kafka_partition INT, kafka_offset BIGINT,
  violations      ARRAY<STRING>,         -- reason codes
  pipeline_version STRING,
  quarantined_at  TIMESTAMP,
  status          STRING,                -- NEW | REPAIRED | DISCARDED | REPLAYED
  repaired_payload STRING,
  reviewed_by     STRING
) USING DELTA CLUSTER BY (quarantined_at);

• Keep the original payload and a pointer to Bronze, so any record is traceable and repairable.
• Repair and replay path: after a fix (for example a mapping was added for a new status string, or a bug in normalization was corrected), a repair job selects rows with status = 'REPAIRED' (or by reason code), pushes them through the same Silver logic, and marks them REPLAYED. Because Silver is idempotent, replay cannot create duplicates.
• Alert on quarantine rate and on reason-code spikes. A sudden jump of UNKNOWN_EVENT_TYPE or CORRUPT_PAYLOAD usually means a firmware or gateway change, not random noise.
• Quarantine is not for late events. Late data is valid data; it flows through normally.

5.4 Deduplication: two different problems

Technical duplicates: the same event_id delivered more than once (broker redelivery, gateway or charge-point retry after a missing acknowledgement). Business duplicates: different event_ids describing the same real-world fact (a charge point resends a reading under a new message ID, or two gateway nodes emit the same event).

Business key definition, per event type (agree this with the business):

Compute business_key = sha2(concat_ws('|', ...normalized key fields...), 256) in Silver.

Resolution rules

1. Drop exact event_id repeats: keep the first by (ingest_ts, kafka_offset).
2. For business-key repeats: keep one deterministic winner (earliest by event_ts, then ingest_ts, then event_id) and write the others to duplicate_events_audit with duplicate_type and the winner's event_id.
3. Conflicting business duplicates (same key but a different meter value): this is not a harmless duplicate. Keep neither silently; flag as CONFLICT, quarantine the later one for review, and alert. Treating conflicts as duplicates is how wrong energy numbers sneak in.

Where deduplication runs: a foreachBatch step that dedupes within the batch, anti-joins against recent Silver, and writes with an insert-only MERGE:

def silver_batch(batch_df, batch_id):
    parsed = normalize_and_validate(batch_df)            # adds is_valid, violations, business_key
    write_quarantine(parsed.filter("NOT is_valid"))      # MERGE on quarantine_id (idempotent)

    valid = parsed.filter("is_valid")

    # within-batch dedup (technical, then business)
    valid = dedup_within_batch(valid, "event_id",     order=["_ingest_ts", "kafka_offset"])
    valid = dedup_within_batch(valid, "business_key", order=["event_ts", "_ingest_ts", "event_id"])

    # against existing Silver, restricted to a date window so the join/merge prunes
    lo, hi = date_bounds(valid, pad_days=1)
    valid.createOrReplaceTempView("incoming")
    spark.sql(f"""
      MERGE INTO silver.charging_events t
      USING incoming s
        ON t.business_key = s.business_key
       AND t.event_date BETWEEN DATE'{lo}' AND DATE'{hi}'     -- prune
      WHEN NOT MATCHED THEN INSERT *                           -- insert-only: replay-safe
    """)
    log_removed_duplicates_to_audit(...)                       # rows that matched = duplicates

Design reasoning, and the alternative

• Insert-only MERGE gives a global duplicate guarantee backed by durable storage, and is safe under replay (re-running the same batch inserts nothing new).
• The date-window predicate is essential for cost: without it the MERGE scans the whole table. The pad accounts for late events; very late duplicates beyond the window are caught by the nightly reconciliation (Section 8).
• Alternative: Structured Streaming dropDuplicatesWithinWatermark("event_id") [verify availability]. It is cheap but only guarantees dedup within the watermark window and holds state; fine as an additional first filter, not as the only defense.
• A separate event_id registry is unnecessary at this volume; the event_id check is the anti-join in the same step.

5.5 silver.charging_events table

CREATE TABLE silver.charging_events (
  event_id          STRING NOT NULL,
  business_key      STRING NOT NULL,
  charge_point_id   STRING NOT NULL,
  connector_id      INT,
  event_type        STRING NOT NULL,
  event_ts          TIMESTAMP NOT NULL,        -- event time, UTC
  event_date        DATE GENERATED ALWAYS AS (CAST(event_ts AS DATE)),
  ingest_ts         TIMESTAMP NOT NULL,        -- arrival time
  transaction_id    BIGINT,
  meter_wh          DECIMAL(18,3),             -- canonical unit, cumulative
  connector_status  STRING,
  stop_reason       STRING,
  ocpp_version      STRING,
  schema_version    INT,
  extra_fields      MAP<STRING,STRING>,        -- rescued/unknown fields
  bronze_topic STRING, bronze_partition INT, bronze_offset BIGINT   -- lineage
)
USING DELTA
CLUSTER BY (event_date, charge_point_id, transaction_id)
TBLPROPERTIES (
  'delta.enableChangeDataFeed'       = 'true',
  'delta.enableDeletionVectors'      = 'true',
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.columnMapping.mode'         = 'name'
);
ALTER TABLE silver.charging_events ADD CONSTRAINT valid_meter CHECK (meter_wh IS NULL OR meter_wh >= 0);

• Event-time grain and append-only: this table never "decides" that an event is too late.
• Clustering on (event_date, charge_point_id, transaction_id) serves the two hot access patterns: date-pruned MERGE for dedup, and "fetch all events of these transactions" for session recompute. Liquid clustering is preferred over static partitioning plus Z-order because the clustering keys can be changed without rewriting the design, and it handles skew better [tune against real query patterns].
• NOT NULL and CHECK constraints are the last line of defense: even if validation logic regresses, a bad row fails loudly instead of corrupting data.
• CDF enabled so downstream jobs can read exactly which events were added per commit.
• Column mapping by name allows renames and drops without rewriting files.
• PII: idTag (user identity) is hashed or tokenized here, with raw values restricted in Bronze via Unity Catalog permissions and masking; deletion requests follow the erase-while-keeping-financial-facts pattern from the earlier GDPR discussion.

6. OUT-OF-ORDER AND LATE EVENTS: THE CORE PRINCIPLE

Make every derived value a pure function of the full set of events, not of arrival order. Then late and out-of-order data stop being special cases.

Concretely:

• Silver uses event time and never rejects late data. An event arriving 3 days late is inserted like any other.
• Watermarks are used only to bound streaming state (if I use state for dedup), never to filter business data. I explain this explicitly because "watermark drops late data" is the most common misunderstanding in streaming interviews.
• Ordering inside a transaction uses event_ts, with tie-breakers: transaction seqNo where OCPP 2.0.1 provides it, otherwise meter_wh (cumulative, so it is monotonic) and ingest_ts.
• Gold derives from Silver by recomputing affected keys (Section 7). A late MeterValues, or a StopTransaction that arrives before its StartTransaction, simply marks the session as affected and the recompute produces the correct result.
• Real-world source of lateness: OCPP 1.6 charge points queue transaction messages while offline and flush them on reconnect, so late and duplicated events are expected, not exceptional. Measure the actual late-arrival lag (ingest_ts - event_ts percentiles) and use it to set the recompute window, the reconciliation horizon and the abandoned-session timeout.

7. GOLD LAYER

7.1 gold.charging_sessions: one row per session

Session identity: session_id = sha2(charge_point_id || '|' || transaction_id). Transaction IDs are not guaranteed globally unique across charge points, so the composite key is safer.

CREATE TABLE gold.charging_sessions (
  session_id          STRING NOT NULL,
  charge_point_id     STRING NOT NULL,
  connector_id        INT,
  transaction_id      BIGINT NOT NULL,
  start_ts            TIMESTAMP,
  stop_ts             TIMESTAMP,
  last_event_ts       TIMESTAMP,
  duration_seconds    BIGINT,
  meter_start_wh      DECIMAL(18,3),
  meter_stop_wh       DECIMAL(18,3),
  last_meter_wh       DECIMAL(18,3),
  energy_kwh          DECIMAL(18,3),
  energy_is_final     BOOLEAN,                -- true only when start and stop readings both exist
  session_status      STRING,                 -- IN_PROGRESS | COMPLETED | TIMED_OUT | START_MISSING
  stop_reason         STRING,
  event_count         INT,
  quality_flags       ARRAY<STRING>,          -- START_MISSING, STOP_MISSING, METER_ANOMALY, CONFLICT...
  row_hash            STRING,                 -- hash of business columns, to skip no-op updates
  updated_at          TIMESTAMP
)
USING DELTA CLUSTER BY (charge_point_id, start_ts)
TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

7.2 Energy calculation and handling incomplete sessions

Meter readings are cumulative, so:

final energy       = meter_stop − meter_start               (needs Start and Stop)
provisional energy = last_meter_reading − meter_start        (session still open)

• energy_is_final = TRUE only when both readings exist. Billing and official reporting use final rows only; live dashboards may show provisional ones, clearly labeled.
• Sessions are never hidden because they are incomplete. They appear immediately as IN_PROGRESS with growing provisional energy.
• State machine

              StartTransaction
 (nothing) ───────────────────────► IN_PROGRESS ──StopTransaction──► COMPLETED (final)
     │                                   │
     │ Meter/Stop arrives first          │ no events for 24h (sweeper)
     ▼                                   ▼
 START_MISSING ──Start arrives──► (recompute) ...        TIMED_OUT (energy stays provisional)

• START_MISSING: Stop or MeterValues arrived before the Start (out-of-order or lost). The session row is created with a flag; when the Start arrives, the recompute repairs it.
• TIMED_OUT: no stop after a configurable period. Evaluated by a periodic sweeper job over the small set of open sessions, using the pipeline's data clock (not now() inside the recompute) so the recompute stays deterministic. If a late StopTransaction arrives afterward, the session flips to COMPLETED and becomes final.
• Meter anomalies: meter_stop < meter_start (meter reset, replacement, rollover or bad data). Do not emit negative energy. Set energy_kwh NULL (or compute the sum of positive deltas between consecutive readings as a documented fallback), add METER_ANOMALY, and surface it in data-quality dashboards. Also cross-check: a stop reading lower than the last MeterValues reading is flagged.
• Units were normalized in Silver, so /1000 is the only conversion here.

7.3 Recompute-affected-keys pattern

def gold_sessions_batch(batch_df, batch_id):
    # batch_df: newly inserted Silver events (append-only source, or CDF)
    affected = (batch_df.filter("transaction_id IS NOT NULL AND event_type IN ('StartTransaction','MeterValues','StopTransaction')")
                        .select("charge_point_id", "transaction_id").distinct())
    affected.createOrReplaceTempView("affected")

    recomputed = spark.sql("""
      SELECT
        sha2(concat_ws('|', e.charge_point_id, CAST(e.transaction_id AS STRING)), 256) AS session_id,
        e.charge_point_id, e.transaction_id,
        max(e.connector_id)                                                             AS connector_id,
        min(e.event_ts) FILTER (WHERE e.event_type = 'StartTransaction')                AS start_ts,
        max(e.event_ts) FILTER (WHERE e.event_type = 'StopTransaction')                 AS stop_ts,
        max(e.event_ts)                                                                 AS last_event_ts,
        min_by(e.meter_wh, e.event_ts) FILTER (WHERE e.event_type = 'StartTransaction') AS meter_start_wh,
        max_by(e.meter_wh, e.event_ts) FILTER (WHERE e.event_type = 'StopTransaction')  AS meter_stop_wh,
        max_by(e.meter_wh, e.event_ts) FILTER (WHERE e.meter_wh IS NOT NULL)            AS last_meter_wh,
        count(*)                                                                        AS event_count
      FROM silver.charging_events e
      JOIN affected a USING (charge_point_id, transaction_id)       -- clustering makes this selective
      WHERE e.event_type IN ('StartTransaction','MeterValues','StopTransaction')
      GROUP BY e.charge_point_id, e.transaction_id
    """)
    final_df = derive_status_energy_flags(recomputed)     # energy_kwh, energy_is_final, status, flags, row_hash
    final_df.createOrReplaceTempView("recomputed")

    spark.sql("""
      MERGE INTO gold.charging_sessions t
      USING recomputed s ON t.session_id = s.session_id
      WHEN MATCHED AND t.row_hash <> s.row_hash THEN UPDATE SET *     -- skip no-op rewrites
      WHEN NOT MATCHED THEN INSERT *
    """)

Why recompute instead of stateful aggregation

• It is stateless and idempotent: the result depends only on Silver contents, so replays, retries and duplicates cannot change the outcome.
• It handles late and out-of-order data without special code.
• The cost is one clustered read per affected session per micro-batch, small at \~700 events/s because only sessions touched in the batch are recomputed.
• Alternative: applyInPandasWithState/transformWithState per session with timeouts. It avoids re-reading but holds state for every open session, makes late data after state expiry awkward, and complicates replays. I would only choose it at much higher volumes or for sub-second latency.
• Alternative: define the session aggregation as a materialized view in a declarative pipeline, so Databricks maintains it incrementally [verify incremental refresh support for the query shape]. Simplest to operate; less control over latency and cost than foreachBatch.

7.4 Other Gold tables

• gold.charging_stats_daily (grain: charge point × day): sessions started, completed, timed-out; energy_kwh_final, energy_kwh_provisional as separate columns (never silently mixed); average duration; utilization. A session belongs to the day of start_ts (documented decision; energy for midnight-spanning sessions can be split by interval if the business needs it). Because late events change past days, the job recomputes the dates touched by changed sessions (found via Gold's CDF) and overwrites those date partitions, an idempotent replaceWhere/MERGE.
• gold.connector_status_history: StatusNotification events as intervals (valid_from, valid_to, status), with valid_to from LEAD(event_ts) per connector, recomputed per affected connector for late events. gold.connector_current_status: latest state via MERGE with WHEN MATCHED AND s.event_ts > t.event_ts, so older or duplicate events can't overwrite newer state. (Declarative pipelines' AUTO CDC / APPLY CHANGES with sequence_by = event_ts implements this pattern natively [verify naming].)
• gold.charge_point_health: last Heartbeat/any event per charge point (liveness), plus offline detection (no event for N minutes).
• StatusNotification events not tied to sessions (from the sample) never enter the sessions logic: the session recompute filters by event type and transaction_id, and status events flow to the status tables.
• Heartbeats: high volume, low analytical value; reduce to last_seen per charge point and keep raw rows only in Bronze with shorter retention.
• Optional silver/gold.meter_intervals: energy per interval from consecutive readings (LAG), for power curves; recomputed per affected transaction.

8. STREAMING VS BATCH

Cost control: if "near real-time" means minutes, use trigger(availableNow=True) on a schedule or a 1-5 minute trigger on right-sized job clusters or serverless instead of an always-on large cluster. State the SLA-versus-cost trade-off: seconds-level latency costs far more than 5 minutes and is rarely needed for session analytics.

One code path: the batch reconciliation reuses the same transformation functions as streaming, so there is no separate "batch logic" to drift (a Kappa-style approach where batch is just a bounded replay).

9. SCHEMA EVOLUTION STRATEGY

1. Bronze is schema-agnostic (raw string/VARIANT + metadata), so upstream changes can never break ingestion.
2. Silver parses with explicit, versioned schemas per event type and ocpp_version. Unknown fields are kept in _rescued_data/extra_fields, not lost, and the rescued-data rate is monitored; an increase triggers a schema review.
3. Additive changes (new nullable field): reviewed, then ALTER TABLE ADD COLUMN, update the parser, and replay from Bronze to backfill the new column when needed.
4. Breaking changes (type change, rename, removed field): schema_version routes to the right parser; unparsable versions go to quarantine with SCHEMA_MISMATCH instead of crashing the stream. Renames cannot be inferred automatically; they need an explicit mapping step.
5. Gold changes only through reviewed pipeline changes, never through automatic mergeSchema, so curated contracts don't change silently. Delta column mapping (name) allows renames and drops safely.
6. Prevention: a schema registry or a documented data contract with the gateway/firmware team (versioned schema, deprecation notice, compatibility rules).

10. IDEMPOTENCY AND FAILURE SCENARIOS

Idempotency per layer

11. DELTA LAKE FEATURES USED, AND WHY

12. WALK-THROUGH ON THE SAMPLE DATA

Dataset (times simplified; ingest_ts is arrival):

Outcomes:

• Row 3: removed as a technical duplicate; duplicate_events_audit records TECHNICAL, pointing at e2.
• Row 4: same business key as row 2 → removed as BUSINESS duplicate, audited. Had its meter value differed, it would be flagged CONFLICT and quarantined instead.
• Row 7: charge_point_id NULL. Recovery from the Kafka key is attempted; if absent, the row is quarantined with MISSING_CHARGE_POINT_ID, with the raw payload retained for repair/replay.
• Row 8: valid; goes to Silver and the status tables; never touches sessions.
• Row 9: valid; updates charge_point_health.last_seen; no session impact.
• Session 501 evolution in Gold:

If the Stop had not arrived, the late reading (7200 at 10:15) would have raised provisional energy to 6.200 kWh on recompute.

• Reconciliation (conservation law): Bronze rows = Silver valid + duplicate audit + quarantine for any time window. Here: 9 Bronze rows = 6 in Silver (e1, e2, e5, e4, e7, e8) + 2 in the duplicate audit (rows 3 and 4) + 1 in quarantine (row 7). The check must hold exactly; a mismatch means a record was lost or double-counted.

13. MONITORING, DATA QUALITY AND RECONCILIATION

Metrics (per run and rolling)

• Freshness: now - max(event ingest) at each layer; end-to-end event-to-Gold latency (p50/p95).
• Volume vs baseline per weekday/hour; consumer lag on the broker.
• Quarantine rate overall and by reason code; alert on thresholds and on new reason codes.
• Duplicate rates (technical and business) and conflict count (should be near zero).
• Late-arrival lag: ingest_ts - event_ts percentiles and the share of events later than the reconciliation window.
• Rescued-data / unknown-field rate (schema drift early warning).
• Session health: open sessions older than expected, START_MISSING count, TIMED_OUT count, METER_ANOMALY count, share of sessions with energy_is_final = false by age.
• Row accounting: the Bronze = Silver + duplicates + quarantine check, per hour.
• Business sanity: energy per session within plausible bounds (for example above a physical maximum for the connector rating), duration plausibility.

Operational design: structured logs with run_id/batch_id; alerts to on-call for freshness SLO breach, accounting mismatch and quarantine spikes (symptom-based, with runbook links); everything also lands in a dq_metrics table feeding a dashboard. Declare data SLOs per Gold table.

Nightly reconciliation recomputes sessions and daily stats for the last N days from Silver, compares to Gold (differences are logged, then repaired), re-checks duplicates beyond the merge window, and verifies accounting.

14. TRADE-OFFS, ALTERNATIVES AND LIKELY FOLLOW-UPS

Q: Why not stateful streaming for sessions? A: Recompute-by-key is stateless, idempotent and tolerant of unbounded lateness, at the cost of re-reading each affected session's events. At this volume that's cheap. Stateful aggregation wins only for much higher throughput or sub-second latency, and it complicates replay and late data after state expiry.

Q: Why not use event_id alone for deduplication? A: It only handles retries. Business duplicates carry new IDs, and without the business key they would double-count meter readings and distort session energy and event counts.

Q: How long can events be late, and what do you do beyond that window? A: Silver accepts any lateness. The merge dedup window and the sessions recompute are keyed by transaction, so very late events still work; the nightly reconciliation covers dedup beyond the pruned window. Measure the late-lag distribution and set horizons from data, not guesses.

Q: What if a charge point's clock is wrong? A: Event time comes from the device, so bad clocks create future or ancient timestamps. Validate against broker_ts/ingest_ts with a tolerance, flag skew, and (policy decision) either quarantine or keep the event with a CLOCK_SKEW flag and use ingest time for ordering fallback. Track skew per charge point to find faulty devices.

Q: How do you test this pipeline? A: Unit tests for pure functions (normalization, validation, business-key, energy derivation); a golden dataset like the sample above with expected outputs for duplicates, late, out-of-order, corrupted, and status-only cases, run in CI; property-style checks (shuffle the input order and replay twice: Gold must be identical); a dev catalog with replay of production samples; and the reconciliation checks as automated assertions.

Q: How would you scale 10x? A: More broker partitions and Spark parallelism, larger micro-batches (maxOffsetsPerTrigger), compaction and clustering tuned on measured query patterns, Heartbeats aggregated at the gateway, and (only if latency demands) stateful session processing.

Q: How do you deploy it? A: Databricks Asset Bundles in Git, with dev/test/prod targets and separate catalogs; schema changes through reviewed migrations; replay/backfill runbooks; same code path for streaming and reconciliation.

Q: What would you change first if requirements tighten to billing-grade accuracy? A: Treat only energy_is_final = true as billable, add stricter conflict handling (no automatic winner; manual review), keep an immutable audit trail of every recompute (Delta history plus a session-version table), validate meter readings against the charge point's reported totals, and require cross-checks with the billing system via reconciliation.

15. CHEAT SHEET

Closing line: "The design rests on one idea: derived tables are deterministic functions of an append-only, deduplicated event log. If Silver is correct and replayable, then late data, out-of-order data, retries, schema changes and bug fixes all reduce to the same operation: recompute the affected keys and merge."