Case Study: A Real Near-Real-Time Lakehouse Mapped to OCPP

The OCPP design from Part 11 mapped onto a working near-real-time lakehouse on Databricks: what such a pipeline typically has, what OCPP adds, and the lessons it teaches.

Part 11 gave a model answer from first principles. This part shows the stronger interview move: anchoring the same design in a pipeline that actually ran. The example is a near-real-time Databricks lakehouse over a public websocket feed of token-trading events (a crypto launchpad), not EV charging. The shape is the same: near-real-time JSON events, many event types, and entity state derived from many events. So the answer becomes "I built this exact pattern on a different feed; here is how it maps."

The case study is honest about what such a pipeline has and does not have. Labels:

When you use your own project this way, keep the same discipline: say "built" only for what exists, and "I would add" for the rest.

1. WHY THE ANALOGY WORKS

2. THE SPOKEN ANSWER (about 4 minutes)

"I'll answer from a pipeline I built with the same shape: a near-real-time feed of heterogeneous JSON events on Databricks, then map it to OCPP and call out what OCPP needs on top.

What I built. The source was a websocket feed of token-trading events. An always-on producer buffers events into batches, compresses them, and lands them as files in a Unity Catalog Volume. A single Lakeflow Declarative Pipeline reads them with Auto Loader into Bronze, which stores each event as raw JSON text, because the payload shape depends on the event type and I didn't want inference or evolution rules deciding what's allowed to land. The same pipeline fans out into Silver tables per event family, with typed columns, array fields exploded into one row per element, and data-quality expectations. Gold is a separate task: a foreachBatch job that streams the entity keys touched by new Silver rows, recomputes only those entities from full Silver history, and MERGEs them. It's deployed with Asset Bundles as a job that triggers the pipeline and then the Gold task.

Mapping to OCPP. Same layers. Bronze keeps the raw OCPP JSON plus broker coordinates. Silver has transaction events (Start/Meter/Stop), status events and heartbeats as separate tables. The charging session is the equivalent of my token entity: the Gold sessions table is a deterministic function of all events for a transaction, recomputed for the transactions touched in each micro-batch. That is why late and out-of-order events are not special cases: a late MeterValues just marks its session as touched and the recompute includes it. Energy is meter_stop − meter_start once the StopTransaction exists. Until then it's provisional from the last reading, flagged energy_is_final = false, and the session shows as IN_PROGRESS.

What OCPP needs that my build doesn't have yet. My Silver layer has expectations but no deduplication and no quarantine, because that feed is append-only and I treated those as follow-ups. For OCPP I'd add, in Silver: a valid/quarantine split with reason codes, technical dedup on event ID and business dedup on a business-key hash, with duplicates audited rather than dropped, and unit and text normalization. The schema strategy stays the same: raw payload in Bronze, versioned parsers in Silver with an extra-fields column for unknown keys, and reviewed changes in Gold.

Lessons that shaped the design. I hit an out-of-memory failure in Bronze decoding that three batch-size settings did not fix; the root cause was Python worker memory not being returned, and the fix was changing the decode approach. I found a silent correctness bug in Gold where a signal read 0.5–16% when the true value was above 99.9%, because it only looked at one of two source tables. And I learned that an incrementally recomputed table can freeze stale rows when logic changes, so I stamp a version on every row. All three are reasons I'd put a reconciliation check, row accounting and a version column into the OCPP design from day one."

3. LAYER BY LAYER

3.1 Producer and landing

In the example:

• ✅ A standalone always-on service (websocket with reconnect), not a Databricks job: a run-then-terminate job is a poor fit for a persistent socket.
• ✅ The writer buffers events (batch of 5000, plus a periodic flush), serializes JSONL, zstd-compresses and base64-encodes into one JSON envelope: {_source, _ingested_at, event_count, codec, uncompressed_bytes, data_b64}.
• ✅ Atomic file publish: write to .tmp, then rename to .json, so a consumer never sees a half-written file. On a write failure the records are put back into the buffer.
• ✅ Idempotent upload: the uploader PUTs to the Files API with 8 workers and treats HTTP 409 ALREADY_EXISTS as success, then moves the file to uploaded/. A re-upload after a crash is harmless.
• ⚠ Known limitation: the buffer is in memory until the flush, so a hard crash loses up to one batch. Acceptable for a public feed; not acceptable for billing-grade OCPP data, which is one reason a durable broker sits in front of Databricks in the OCPP design.

OCPP mapping: the gateway and broker replace this layer. Event Hubs/Kafka gives durable buffering, per-key ordering (key = charge_point_id) and replay, so the envelope/compression step goes away and Bronze reads with readStream.format("kafka").

3.2 Bronze

In the example:

@dp.table(
    name="workspace.bronze.events_raw",
    comment="Raw feed events, decoded from the zstd+base64 batch envelopes ...",
    partition_cols=["_ingested_date"],
)
def events_raw():
    envelopes = (spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.maxFilesPerTrigger", str(MAX_FILES_PER_TRIGGER))   # 20
        .schema(_ENVELOPE_SCHEMA).load(SOURCE_PATH))
    lines = envelopes.select("data_b64", "codec").mapInPandas(_decode_batch, schema=_LINES_SCHEMA)
    return (lines
        .withColumn("_source", F.get_json_object("line", "$._source"))
        .withColumn("_ingested_at", F.to_timestamp(F.get_json_object("line", "$._ingested_at")))
        .withColumn("_ingested_date", F.to_date("_ingested_at"))
        .withColumn("event", F.get_json_object("line", "$.event"))
        .select("_source", "_ingested_at", "_ingested_date", "event"))

• ✅ Raw JSON text in Bronze, not a parsed struct: the payload structure varies per action, so schema inference/evolution is a poor fit. The same argument holds for OCPP, where MeterValues and StatusNotification have different shapes and OCPP 1.6 and 2.0.1 differ.
• ✅ Fixed, known envelope schema (no inference needed because the producer controls that shape).
• ✅ Partitioned by _ingested_date (arrival date, not event date, so late data doesn't scatter writes).
• ✅ Auto Loader tracks processed files in its checkpoint, so each landed file is ingested once; combined with the idempotent upload that gives file-level idempotency.
• ✅ Scale: a full triggered update processed about 193.5M rows into Bronze without error after the decode fix.
• ⚠ Gaps worth knowing:
  • Bronze keeps no pointer back to the source file or offset (no _metadata.file_path), so a bad row can't be traced to its file. OCPP version: keep (topic, partition, offset).
  • In _decode_batch, rows with a missing or unsupported codec are skipped silently. In the OCPP design this becomes a quarantine row with a reason code, since silent loss breaks the row-accounting check.

The OOM story (the best "scar" in this layer): a scalar UDF that decompressed whole envelopes crashed with UDF_PYSPARK_ERROR.OOM. Three batch-size knobs (Auto Loader maxFilesPerTrigger 100 then 20, then Arrow maxRecordsPerBatch=5) all failed at about the same wall-clock mark, which pointed to the reused Python worker never returning memory to the OS rather than any single batch being too large. spark.python.worker.reuse=false was the direct fix but serverless pipelines reject it. The resolution was rewriting the decode as mapInPandas (a generator that yields per batch), explicitly dropping buffers and calling gc.collect() after each batch. How to say it: "I stopped tuning symptoms once three different batch sizes failed at the same point; identical failure time across settings meant the cause wasn't batch size."

OCPP extension ➕ (Bronze reader only; everything downstream is unchanged):

@dp.table(name="workspace.bronze.raw_ocpp_events", partition_cols=["_ingested_date"])
def raw_ocpp_events():
    return (spark.readStream.format("kafka").options(**kafka_opts)
        .option("subscribe", "ocpp-events").option("maxOffsetsPerTrigger", 500_000).load()
        .select(F.col("value").cast("string").alias("event"),
                F.col("key").cast("string").alias("_kafka_key"),
                "topic", "partition", "offset",
                F.current_timestamp().alias("_ingested_at"))
        .withColumn("_ingested_date", F.to_date("_ingested_at")))

3.3 Silver

In the example:

• ✅ One Silver table per event family, all reading spark.readStream.table(BRONZE_TABLE): events (every event, fixed schema via from_json, _raw_event kept), tokens (create), transfers, pools, trades.
• ✅ Array explosion with a technical key: posexplode(breakdown) gives one row per trade with (signature, breakdown_index); posexplode(transfers) gives (signature, transfer_index). signature alone is not unique because one transaction can emit several events. That is the same insight as "business duplicates need a business key."
• ✅ Event time is a typed column (event_time) derived from the payload timestamp, kept separate from _ingested_at (arrival time). This separation is what makes late-data reasoning possible.
• 🟡 Data quality: @dp.expect("valid_signature", "signature IS NOT NULL") and similar. These record violation metrics but keep the rows. They are monitoring, not enforcement.
• ⚠ Not implemented: deduplication (keys documented, not enforced), quarantine, rescued/unknown-field capture (the schemas are fixed, so new fields are ignored rather than captured; the raw event is still retained in Bronze).

OCPP extension ➕ (Silver additions in the same declarative style):

1. Valid/quarantine split. Use the same dp.expect* machinery, but drop-and-route instead of warn:

def _nn(cond):                      # null-safe: a NULL rule result counts as a violation
    return f"COALESCE(({cond}), FALSE)"

RULES = {
  "MISSING_EVENT_ID":        _nn("event_id IS NOT NULL"),
  "MISSING_CHARGE_POINT_ID": _nn("charge_point_id IS NOT NULL"),
  "UNKNOWN_EVENT_TYPE":      _nn("event_type IN ('StartTransaction','MeterValues','StopTransaction','StatusNotification','Heartbeat')"),
  "INVALID_EVENT_TS":        _nn("event_ts IS NOT NULL AND event_ts <= current_timestamp() + INTERVAL 5 MINUTES"),
  "MISSING_TRANSACTION_ID":  _nn("event_type IN ('StatusNotification','Heartbeat') OR transaction_id IS NOT NULL"),
  "INVALID_METER_VALUE":     _nn("event_type <> 'MeterValues' OR meter_wh >= 0"),
}

@dp.table(name="ocpp_events_valid")
@dp.expect_all_or_drop(RULES)                       # [verify API name in your runtime]
def ocpp_events_valid():
    return _parse_and_normalize(spark.readStream.table(BRONZE))   # plain function, shared

@dp.table(name="ocpp_events_quarantine")
def ocpp_events_quarantine():
    p = _parse_and_normalize(spark.readStream.table(BRONZE))
    failed = F.filter(F.array(*[F.when(~F.expr(r), F.lit(k)) for k, r in RULES.items()]),
                      lambda x: x.isNotNull())
    return p.filter(F.size(failed) > 0).withColumn("violations", failed)   # + raw payload + Bronze coordinates

_parse_and_normalize also does the safe deterministic fixes (trim/case, unit normalization Wh→canonical, recover charge_point_id from the Kafka key before quarantining) and computes business_key.

2. Deduplication needs a MERGE, which @dp.table can't express. So it uses the same escape hatch as Gold: a foreachBatch task with an insert-only merge, pruned by date:

def dedup_insert(batch_df, batch_id):
    _spark = batch_df.sparkSession            # Spark Connect convention
    w = Window.partitionBy("business_key").orderBy("event_ts", "_ingested_at", "event_id")
    deduped = batch_df.withColumn("_rn", F.row_number().over(w)).filter("_rn = 1").drop("_rn")
    lo, hi = deduped.agg(F.min("event_date"), F.max("event_date")).first()
    (DeltaTable.forName(_spark, SILVER_EVENTS).alias("t")
        .merge(deduped.alias("s"),
               f"t.business_key = s.business_key AND t.event_date BETWEEN '{lo}' AND '{hi}'")
        .whenNotMatchedInsertAll().execute())          # insert-only: replay-safe

Rows that matched go to a duplicate_events_audit table (TECHNICAL for repeated event_id, BUSINESS for the same key with a new ID, CONFLICT for the same key with different values, which is quarantined instead of merged). Alternative: an AUTO CDC flow (formerly APPLY CHANGES) keyed on business_key with sequence_by = event_ts [verify exact API name]; it is declarative but gives less control over conflict handling.

3. Business key per event type (agree with the business): Start = (charge_point, connector, transaction); Stop = (charge_point, transaction); MeterValues = (charge_point, transaction, event_ts, measurand); Status = (charge_point, connector, status, event_ts).

3.4 Gold: the strongest part

In the example:

touched_tokens_stream = (
    spark.readStream.table(TOKENS_TABLE).select("token_id")
    .unionByName(spark.readStream.table(POOLS_TABLE).select("token_id"))
    .unionByName(spark.readStream.table(TRADES_TABLE).select("token_id"))
    .unionByName(spark.readStream.table(TRANSFERS_TABLE).select("token_id")))

def upsert_gold_risk(batch_df, batch_id):
    _spark = batch_df.sparkSession
    touched = batch_df.select("token_id").distinct()
    result = compute_risk_for_tokens(_spark, touched)     # reads full Silver history for those tokens
    (DeltaTable.forName(_spark, GOLD_TABLE).alias("t")
        .merge(result.alias("s"), "t.token_id = s.token_id")
        .withSchemaEvolution().whenMatchedUpdateAll().whenNotMatchedInsertAll().execute())

(touched_tokens_stream.writeStream.foreachBatch(upsert_gold_risk)
    .option("checkpointLocation", CHECKPOINT_PATH).trigger(availableNow=True).start())

• ✅ Recompute-by-touched-key + MERGE. The streaming query only identifies which entities changed; the actual derivation reads the entity's full history from Silver in batch mode.
• ✅ Why it isn't a declarative table: the scoring relies on row_number/rank/window aggregates, and arbitrary non-time-based windows aren't supported on streaming DataFrames; inside foreachBatch each micro-batch is a plain batch DataFrame so windows work. Also a materialized view and a MERGE-written table can't both own the same table.
• ✅ Order-independence: because each recompute uses windows over all of an entity's rows (earliest create row, latest pool state, ranking), arrival order doesn't change the result. This is exactly the property the OCPP session needs for late and out-of-order events.
• ✅ Versioning: SCORING_MODEL_VERSION stamped on every row, with a documented procedure: after a logic change, reset the checkpoint to force a full re-score.
• ✅ Spark Connect convention (batch_df.sparkSession, not the global spark) and .withSchemaEvolution() instead of the session conf that serverless rejects.

OCPP adaptation ➕:

touched_txn_stream = (spark.readStream.table("workspace.silver.ocpp_transaction_events")
                           .select("charge_point_id", "transaction_id"))

def upsert_sessions(batch_df, batch_id):
    _spark = batch_df.sparkSession
    keys = batch_df.dropna().distinct()
    result = compute_sessions_for_keys(_spark, keys)    # aggregation + status/energy logic from Part 11
    (DeltaTable.forName(_spark, GOLD_SESSIONS).alias("t")
        .merge(result.alias("s"), "t.session_id = s.session_id")
        .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute())

Derivation (unchanged from Part 11): energy_kwh = (meter_stop − meter_start)/1000 when both exist (energy_is_final = true); otherwise provisional from the last reading; statuses IN_PROGRESS, COMPLETED, TIMED_OUT, START_MISSING; anomaly flags for negative energy.

A time-driven signal needs a sweeper. A flag defined relative to now (for example "creator sold recently") stays stale when only entities with new activity are recomputed: nothing touches a quiet entity. OCPP has the same issue with TIMED_OUT: an abandoned session produces no new events, so nothing triggers its recompute. The fix is a small scheduled sweeper over open sessions using a data clock, the same class of fix as the version-driven re-score.

3.5 Orchestration and deployment

• ✅ Databricks Asset Bundles: one job with two tasks: pipeline_task (the Lakeflow pipeline) then notebook_task (Gold), depends_on the first. Cron 0 */2 * * * ? (every 2 minutes, UTC).
• ✅ Pipeline settings: serverless, continuous: false (triggered, not always-on), channel: CURRENT, dependency zstandard via the pipeline environment (plain .py files can't use %pip), and spark.sql.execution.arrow.maxRecordsPerBatch: "5" as part of the OOM mitigation.
• ✅ Why one pipeline instead of separate Bronze and Silver jobs: they share one DAG, dependencies are expressed in code (readStream.table(BRONZE_TABLE)) rather than job scheduling, and Databricks manages one cluster/update lifecycle.
• ✅ Why independent of other orchestrators: different domain, different cadence, different failure blast radius.
• ✅ Why triggered every 2 minutes rather than continuous: "near real-time" is minutes here, and triggered runs on serverless cost far less than an always-on stream. For OCPP, state the same trade-off against the freshness SLO.
• Be accurate about operations: if a schedule is paused while you tune, say "scheduled every 2 minutes; currently paused while tuning", not "running every 2 minutes in production".

4. EACH REQUIREMENT: WHAT THE EXAMPLE DOES, AND WHAT TO ADD

5. SCARS AND LESSONS (use these for "what would you change?")

1. Out-of-memory in Bronze decode. Three batch-size settings failed identically, so the cause wasn't batch size; Python worker memory was never returned. Fix: mapInPandas generator, explicit del and gc.collect(); the cleaner fix (worker.reuse=false) was blocked on serverless. Lesson: when different knob settings fail at the same time, stop tuning and find the shared cause.
2. A silent correctness bug in Gold. A holder-concentration signal read only the transfers table, but holders mostly trade against the pool, so across 30 sampled failed tokens it read 0.5–16% when the true trade-based concentration was 99.98–99.996%. A 30-day audit of 15,912 "low risk" tokens surfaced it. Fix: union net positions from trades and transfers. Lesson: nothing failed; only a check against known-bad cases found it. That is the argument for reconciliation and golden datasets (Part 6).
3. Stale rows in an incremental table. 28 of 30 audited tokens had never been re-scored after a new signal shipped because they stopped producing events. Fix: a model version on every row and a forced re-score procedure. Lesson: incremental recompute needs a version column and a replay path.
4. First run is a full backfill. No checkpoint means every existing entity is "new". Lesson: size the first run like a full rebuild.
5. Silent skip of unsupported codec rows. Easy to miss and impossible to reconcile. Lesson: every dropped record needs a destination (quarantine) so row accounting balances.
6. Where declarative pipelines stop. Windows on streaming data and MERGE ownership pushed Gold into foreachBatch. Lesson: know the framework's limits before choosing it for a layer.

6. LIKELY FOLLOW-UPS

Q: Why store the raw event as a JSON string in Bronze? A: The payload shape depends on the event type, and I want Bronze to be immune to schema changes and corrupted fields. Anything that parses badly in Silver can be re-parsed later from the raw text. In OCPP the same holds across event types and protocol versions.

Q: Why files in a Volume and Auto Loader instead of reading the feed directly? A: Databricks can't hold an always-on websocket, and I didn't want a direct socket or tunnel into the workspace. A producer that writes files and an Auto Loader that reads them is a well-understood pattern. With OCPP there's already a broker, so I'd read it directly.

Q: Why compress and base64-encode batches? A: To cut network and storage volume (roughly 4-5x) at the upload cadence. The cost was CPU and, as it turned out, memory pressure in the Python decode step. With a broker I'd skip it.

Q: Why is Gold outside the pipeline? A: Detecting changed entities needs a streaming read, but the scoring needs arbitrary window functions, which streaming doesn't support. foreachBatch turns each micro-batch into a batch DataFrame. A materialized view and a MERGE-written table also can't own the same table.

Q: How do you handle late data? A: Silver never filters by lateness; it appends. Gold recomputes each touched entity from complete history, so a late event changes the result the next time its entity is touched. For sessions I'd add a nightly recompute to heal anything missed and a sweeper for time-driven states.

Q: How would you make this exactly-once? A: I don't claim end-to-end exactly-once. I use at-least-once delivery with idempotent steps: files processed once via the checkpoint, replay-safe MERGE in Gold, and (for OCPP) insert-only dedup in Silver.

Q: What would you change first? A: In order: (1) quarantine with reason codes and Bronze source coordinates, so no record disappears silently; (2) enforced dedup keys; (3) turn warn-only expectations into drop-and-route for critical rules; (4) a sweeper and a nightly reconciliation; (5) a row-accounting check (Bronze = Silver + duplicates + quarantine) and a golden test dataset; (6) a durable broker instead of an in-memory producer buffer.

Q: How do you deploy and operate it? A: Asset Bundles in Git: a job that triggers the pipeline and then the Gold notebook, with the pipeline serverless and triggered. Pausing the schedule while tuning is one setting and a redeploy.

7. USING YOUR OWN PROJECT THIS WAY

• Say "built" only for what exists. For anything else, say "in this design I'd add", and point to related work you have done as proof you know how.
• Close a gap before the interview if you can. Turning a ➕ item into ✅ (for example a quarantine split) is worth more than describing it. Only claim it once it is merged.
• Check what an interviewer might read. Docs whose status lines lag the code, or a paused schedule, invite awkward questions. Refresh them first.
• Know your numbers: batch sizes, cadence, row counts, the before/after of any bug you fixed.

8. CHEAT SHEET

Closing line: "Streaming correctness comes from structure, not tuning: keep raw data untouched, derive everything else as a deterministic function of an append-only event log, recompute only what changed, and make every dropped record land somewhere you can count."