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

How MQTT is used — the actual transfer

Campus meters sit on Schwarz/Neuberger GLT. They do not call us. They publish a small JSON to Solace. We subscribe. That is the only live feed. Everything after that is our copy of the last count.

1. Meter publishes. About once a minute the device sends its lifetime count on topic neuberger/t/sit/neuberger/{building}/meter/{meter}. Body is values[] + glt_values[]. Not kWh for the last minute. The number on the glass, native unit (kWh / MWh / m³). No clock in the JSON.
2. Airflow opens one MQTT session, then closes it. DAG schwarz_mqtt_to_s3_bucket_v2 every 17 minutes on airflow-rc. Connection sc-mqtt: host pssolsch01.messaging.smartcity.hn:8893, user realcube-client, topic neuberger/t/sit/neuberger/#. CONNACK → wait for first telegram → keep the socket open 14 minutes → write status JSON → disconnect. The 3-minute gap is so two collectors never overlap on the same username. The 14 minutes on the Airflow UI is listen time, not parse time.
3. Each telegram becomes one S3 object. Callback on_message in collector.py: zip arrays → one parquet row → PutObject to s3://realcube-integrations/schwarz-group/{Y}/{M}/{D}/{building}/{meter}/raw/message_{Berlin}_{µs}.parquet. Timestamp = Berlin now. That is why a 14-minute window logs 1449 messages and writes 1449 files.
4. Glue copies those rows into Iceberg, then the tiny files disappear. DAG schwarz_s3_parquet_to_iceberg_merger at :03/:21/:39/:57. Job realcube-integration-schwarz-merge (Spark, 2× G.1X, ~4–5 min). Reads today’s prefix, appends schwarz_group.building_counters_v2 on S3 under iceberg/schwarz_group/building_counters_v2/, archives. Athena table building_counters_v2_current is that location. Leftover raw files are only minutes after the last merge (tonight: 100 meters × 11 files = 1100 objects).
5. Nothing in the product reads the tiny files. Loop A and Loop B query Athena last-wins / first-last of a Berlin day. Meter Management and Exposé never see parquet. They see sc-healthcheck and Rails.

Why ~1400 files

It is arithmetic, not a warehouse requirement.

files_per_window  = meters × minutes_listened × telegrams_per_meter_per_minute
                  ≈ 100    × 14               × 1
                  = 1400

live 23:17 UTC    = 1449 writes   (≈ 103 / min; extra topics + jitter)

files_per_hour    ≈ 100 × 60 = 6000 objects
bytes_of_values   ≈ 100 × 60 × ~80 B JSON  ≈ 480 KB of actual counts
bytes_on_S3       ≈ 100 × 60 × 4262 B      ≈ 25 MB of parquet wrappers

One file = one row = one count. Parquet footer, pandas metadata, decimal256 schema turn ~80 bytes of JSON into 4262 bytes. Glue then spends four minutes concatenating them. We need the 1449 counts. We do not need 1449 objects. One parquet for the window is the same 1449 rows.

Airflow returns len(messages_collected) as XCom. That 1449 is “how many PutObject succeeded”, not “how many meters exist”. 100 meters for 14 minutes, plus a few extra publishes, is the whole story.

Why S3 exists here

The Airflow task dies when the 14-minute window ends. RAM is gone. MQTT is not a database: if we disconnect, the next session does not replay last night. So the counts must land on disk before the process exits. S3 is that disk. Three different jobs read three different keys. They never share a process.

How the counts are used

After Iceberg, two Airflow DAGs consume Athena. They do different jobs. Peter 12 Aug: MQTT liveness stays 15-minute. Daily kWh into energy tables is a different loop.

Worked A01 from leftover files: 01:23 count 179 086.90625, 01:35 count 179 091.109375. Twelve minutes of campus electricity ≈ 4.2 kWh. Loop A would compare a similar delta to the meter’s 15-min / 60-min threshold. Loop B at 00:30 would take last count before tonight’s midnight minus last count before last midnight and POST that one number for 28 Aug.

Grafana “MQTT ONLINE/OFFLINE” on Hetzner is a third reader: it probes Solace CONNACK. Airflow ignores a lone OFFLINE if ONLINE appeared in the last 30 min. That probe is not the ingest path. Do not treat it as campus-down.

Debezium + Kafka — would it help?

No for Debezium. Maybe-later for Kafka, and not at this size. Campus rate is ~100 telegrams/min ≈ 1.7 messages/second. Realcube has no Kafka / MSK / Debezium in this path today. Solace already is the broker. We cannot replace it; Schwarz owns it.

Debezium on sc-healthcheck RDS would also be the wrong table. Healthcheck stores last_value after we PATCH it. CDC would see our writes, not the campus telegrams. You still need the Solace subscriber.

If we meant Kafka without Debezium

That is a different design: MQTT connector writes a Kafka topic keyed by meter; a consumer batches to parquet. Log compaction on the key is last-wins. Replay if the collector dies. Several consumers without extra Solace sessions. Those are real Kafka properties. They do not remove S3: Loop B still needs yesterday’s midnight count, so you keep a lake anyway (compacted topic is “current”, not 27 days of A01).

campus        ≈ 100 msg/min = 1.7 msg/s
Kafka/MSK     = cluster + Connect + IAM + pages when brokers die
this job      = last v(m), v(midnight), Δ15, Δ60

Kafka buys    = durable log, replay, fan-out, compaction ≈ last-wins
we already    = S3 mailbox + Athena last-wins
the bug       = 1 PutObject per telegram + Spark
fix           = buffer → 1 parquet/window; same math, no new broker

Use Kafka when a second independent consumer must read the same telegram stream without touching Solace, or when several campuses share one ingest bus. One campus, two DAGs, 100 meters: a compacted Kafka topic is a heavier way to store last-wins than a parquet and row_number().

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.