The telegram is a counter. We store it like a warehouse.

What we do wrong

MQTT is not kWh. It is the meter’s lifetime count. Consumption is a subtraction of two counts at two timestamps. Health is the same subtraction over 15 minutes and 60 minutes. Daily Exposé is midnight minus previous midnight. That is the whole product.

We instead write every telegram as its own S3 object, boot Spark to merge them, keep Iceberg at MaxConcurrentRuns=1, then run two Airflow clocks against that one writer. Iurii’s SNS Max concurrent runs exceeded is that clock collision. It is not campus-down. Skip-if-running is already on the live 3/18 merger. Do not raise MaxConcurrentRuns.

How MQTT actually looks

Subscribe neuberger/t/sit/neuberger/# on pssolsch01.messaging.smartcity.hn:8893, user realcube-client. Topic suffix is /{building}/meter/{meter}. Body is JSON. Collector zips values with glt_values. Every leftover file tonight has exactly one row. The original telegram is therefore length-1 arrays. There is no timestamp inside the JSON. We stamp Berlin now on receive.

{
  "topic": "neuberger/t/sit/neuberger/A-Bau/meter/A01",
  "payload": {
    "values": [
      { "type": "energy", "unit": "kWh", "value": 179086.90625 }
    ],
    "glt_values": [
      { "glt_name": "A01" }
    ]
  },
  "received_at": "2026-08-28T01:23:13.193483+02:00"
}

S3 object that telegram became: schwarz-group/2026/8/28/A-Bau/A01/raw/message_20260828_012313_193483.parquet, 4262 bytes, schema building, glt_name, type, unit, decimal256 value, timestamp. Glue later maps that to Iceberg meter, meter_type, meter_unit, meter_value, timestamp.

A01 is a counter — 13 leftover minutes, live

Glue last succeeded 23:21–23:26 UTC, so leftover raw files are only the minutes after that merge. A01 ticks every 60.0 s. Value only goes up. Minute delta is ~0.31–0.42 kWh. That is load, not a new “energy event”.

Athena on the same meter, August: 28 719 rows, first 2026-08-01 00:00:41.299, last 2026-08-27 23:22:13.230 UTC (qid ca7677b1-…, scanned 20 304 059 bytes). 27 days × 1440 min = 38 880 slots. Coverage ~74 %, close to the 14-minute listen inside a 17-minute DAG slot (14/17 = 82 %) plus merge lag. We already drop ~3 minutes of telegrams every slot by design.

Same minute, other meters (leftover parquet 01:23 Berlin)

A02, D03, T17 reading 0 is campus state (OPEN_ISSUES §6), not ingest loss. Native unit stays on the wire: U1 and I14 and R1 are MWh, A01 is kWh, H06 is m³. DAG multiplies MWh × 1000 on Athena read. Thresholds stay native. That contract is already correct. The waste is the file layout around it.

The algorithms. Three queries. No Spark.

Let v(m, t) be the cumulative count for meter m at receive time t. Payload has no t; t is Berlin now. All product numbers are differences of v.

-- current reading (Iceberg “_current” table)
SELECT * FROM (
  SELECT *,
         row_number() OVER (PARTITION BY building, meter ORDER BY received_at DESC) AS rn
  FROM read_parquet('dt=*/*.parquet')
) WHERE rn = 1;

-- Berlin day consumption
-- SOD = last v with received_at <= Berlin midnight
-- EOD = last v with received_at < next midnight
kWh_day(m, d) = v(m, EOD_d) - v(m, SOD_d)

-- 60-minute window, time-normalized
TE_60(m, t) = (v(m, t) - v(m, t - 60 min)) * (60 min / actual_gap)

Prod already has the first query in reader.py as Athena SQL. Prod already has the second as query_daily_readings (min_by / max_by per Berlin day). Prod does not have the third: it subtracts a rotated pointer that is 75 minutes old and does not scale by gap. That is leftover TE while 15/60 live values are already under the limit.

This pipe vs that pipe

Glue tonight, last eight runs, all SUCCEEDED, schedule :03/:21/:39/:57 only (skip-if-running is holding): 281 s, 247 s, 387 s, 272 s, 270 s, 252 s, 272 s, 274 s. Worker G.1X × 2. The 387 s run is the :57 overlapping the next :03 in wall time — that is why a second clock at :00 used to SNS-fail. v4 0/15 stays paused, not on RC disk.

Hetzner 40 s probe on 27 Aug was a safety window on the shared user, not the architecture. It proved one 5029-byte file holds 100 meters. That page is archive. This page is the algorithm.

Logs (28 Aug morning, no extra MQTT client)

mqtt_status.json

Collector · dag_id=schwarz_mqtt_to_s3_bucket_v2 · run scheduled__2026-08-27T23:17:00+00:00

CONNACK is instant. First telegram waits ~47 s because the meters publish on a one-minute tick, not because the broker is dead. Task wall time is the 14-minute listen plus that wait. That is why Airflow shows ~14–15 min. It is not compute.

Glue · realcube-integration-schwarz-merge · jr_08b5b4c7…

S3 leftover at 01:31 Berlin (after that Glue)

Leftovers on work.koveh.com

The hub still sells the 40 s Hetzner probe as the ingest story. That page is too thin. This page replaces it. Other Schwarz pages are a different layer (Berichte, CO₂, Peter loops). They stay. They were not updated for Glue skip or for the dual I14 topic.

Text for Iurii

Short. Numbers from this snapshot. Ask is: keep MaxConcurrentRuns=1; optional JobRunQueuingEnabled; do not start a second Glue clock. Ingest cutover is our DAG, not his job.

Do not

• Raise Glue MaxConcurrentRuns.
• Copy merger v4 onto airflow-rc (second live DAG).
• Open another long MQTT session on realcube-client.
• Restart Airflow to pick up a file.
• Auto-ACCEPT leftover TE.
• Treat Hetzner as company prod.

Code for the compact lake already exists under /root/realcube/bugfixes/schwarz-sane-collector/schwarz_ingest/lake.py (DuckDB last-wins, start-of-day). Prod collector remains dags/schwarz/mqtt/collector.py until a surgical host deploy.