Skip to content

feat(berlinmod): streaming benchmark harness with corpus-derived parameters (stacks on #8) #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
estebanzimanyi wants to merge 1 commit into from
Closed

feat(berlinmod): streaming benchmark harness with corpus-derived parameters (stacks on #8) #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 98 additions & 0 deletions kafka-streams-app/docs/benchmark.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
# BerlinMOD streaming benchmark (Kafka Streams)

The Kafka-Streams counterpart of the MobilityFlink BerlinMOD benchmark covers
the 27 BerlinMOD-9 × 3-form cells (Q1–Q9 × continuous / windowed / snapshot)
with two harnesses, both over the same corpus and corpus-derived parameters, and
with the spatial predicates evaluating through MEOS (see
[`MEOSBridge`](../src/main/java/berlinmod/MEOSBridge.java)):

- [`BerlinMODBenchmark`](../src/main/java/berlinmod/BerlinMODBenchmark.java) runs
each cell in isolation through a [`TopologyTestDriver`](../src/main/java/berlinmod/BerlinMODTopology.java)
and reads its output cardinality — an in-process **correctness** harness.
- [`EmbeddedBrokerBenchmark`](../src/test/java/berlinmod/EmbeddedBrokerBenchmark.java)
runs each cell as a real [`KafkaStreams`](../src/main/java/berlinmod/BerlinMODTopology.java)
application against an in-process `EmbeddedKafkaCluster` (a genuine
`KafkaServer` over the loopback network) — the **throughput** harness, the
Flink-comparable analog of MobilityFlink's per-cell jobs.

## Corpus and parameters (regular with MobilityFlink)

The corpus is the real BerlinMOD instants (`--csv`, reprojected EPSG:3857→4326
through MEOS by [`BerlinMODCorpus`](../src/main/java/berlinmod/BerlinMODCorpus.java))
or a synthetic corpus. The per-query parameters (point `P`, region box, road
segment, points of interest, target vehicle ids) **and** the window/tick
granularity are derived from the corpus via `BerlinMODCorpus.derive` — the same
mechanism MobilityFlink uses — and threaded through `BerlinMODTopology.build(Params)`,
so the topology auto-scales to the corpus span instead of carrying a fixed
window/tick. The two stream bindings share one mechanism.

## Throughput harness

Each cell runs against its own fresh `EmbeddedKafkaCluster`, the true analog of
MobilityFlink's independent per-cell jobs: the corpus is produced once into the
input topic, the cell runs as a single-threaded `KafkaStreams` application, and
throughput is the events consumed divided by the wall-clock from streams start
until the application has read the whole input topic (its own
`records-consumed-total` metric reaches the input end offset) and its output has
gone idle. The trailing settle time is excluded from the wall; each consumed
record runs the cell's MEOS predicate, so this is the steady-state per-event
processing rate, directly comparable to the MobilityFlink figures.

Run from `kafka-streams-app/`, with an extended libmeos on the loader path and
the test classpath (it carries the embedded broker):

```
CP=$(mvn -q dependency:build-classpath -DincludeScope=test -Dmdep.outputFile=/dev/stdout | tail -1)
LD_LIBRARY_PATH=<libmeos-dir> java -Dorg.slf4j.simpleLogger.defaultLogLevel=warn \
-cp target/classes:target/test-classes:jar/JMEOS.jar:$CP \
berlinmod.EmbeddedBrokerBenchmark --csv <berlinmod_instants.csv> [--max N] [--only Q3-continuous]
```

## Figures

Real BerlinMOD corpus (216,075 instants, 5 vehicles, ~11 days, EPSG:3857),
single-broker `EmbeddedKafkaCluster`, one stream thread per cell, Java 21,
16-core host; libmeos built with `-DMEOS/CBUFFER/NPOINT/POSE/RGEO=ON`.

| Cell | Events in | Output rows | Wall (ms) | Throughput (ev/s) |
|---|---:|---:|---:|---:|
| Q1-continuous | 216075 | 5 | 1899 | 113,785 |
| Q1-snapshot | 216075 | 703 | 1701 | 127,029 |
| Q1-windowed | 216075 | 86 | 1650 | 130,956 |
| Q2-continuous | 216075 | 61170 | 1289 | 167,631 |
| Q2-snapshot | 216075 | 140 | 1683 | 128,388 |
| Q2-windowed | 216075 | 50 | 1704 | 126,806 |
| Q3-continuous | 216075 | 216075 | 4618 | 46,790 |
| Q3-snapshot | 216075 | 97 | 2607 | 82,883 |
| Q3-windowed | 216075 | 50 | 4188 | 51,594 |
| Q4-continuous | 216075 | 62 | 9356 | 23,095 |
| Q4-snapshot | 216075 | 4685 | 10499 | 20,581 |
| Q4-windowed | 216075 | 98 | 9741 | 22,182 |
| Q5-continuous | 216075 | 60577 | 22889 | 9,440 |
| Q5-snapshot | 216075 | 34 | 2006 | 107,715 |
| Q5-windowed | 216075 | 6 | 2896 | 74,612 |
| Q6-continuous | 216075 | 216075 | 8708 | 24,814 |
| Q6-snapshot | 216075 | 698 | 7258 | 29,771 |
| Q6-windowed | 216075 | 203 | 6268 | 34,473 |
| Q7-continuous | 216075 | 5 | 4321 | 50,006 |
| Q7-snapshot | 216075 | 632 | 7487 | 28,860 |
| Q7-windowed | 216075 | 53 | 11625 | 18,587 |
| Q8-continuous | 216075 | 216075 | 4807 | 44,950 |
| Q8-snapshot | 216075 | 281 | 2825 | 76,487 |
| Q8-windowed | 216075 | 77 | 5946 | 36,340 |
| Q9-continuous | 216075 | 107870 | 4561 | 47,375 |
| Q9-snapshot | 216075 | 140 | 2281 | 94,729 |
| Q9-windowed | 216075 | 22 | 2384 | 90,636 |

The per-event MEOS-predicate cells that emit one row per input event
(Q3/Q8/Q9-continuous, 216,075 output rows through `edwithin_tgeo_geo` /
`eintersects_tgeo_geo`) sustain 45,000–47,000 ev/s. In the cross-platform
streaming catalog these cells run below MobilityFlink's in-JVM mini-cluster on
the same corpus, because Kafka Streams routes every record through the broker;
the per-cell shape matches across both engines. Q5-continuous is the O(V²)
all-pairs-meeting outlier (9,440 ev/s). Non-spatial cells (Q1/Q2) reach
110,000–168,000 ev/s. The `TopologyTestDriver` correctness harness
([`BerlinMODBenchmark`](../src/main/java/berlinmod/BerlinMODBenchmark.java))
reports output cardinality and correctness, not throughput: its per-event
bookkeeping dominates wall-clock, so a no-predicate cell already runs far below a
real broker's rate.
54 changes: 54 additions & 0 deletions kafka-streams-app/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,60 @@
<version>${junit.version}</version>
<scope>test</scope>
</dependency>

<!-- Embedded Kafka broker for the runtime throughput benchmark.
EmbeddedKafkaCluster runs a real in-process KafkaServer (over the
loopback network) so BerlinMODBenchmark measures steady-state
throughput against an actual broker rather than the in-memory
TopologyTestDriver. The broker core (kafka_2.13) pulls in
scala-library and ZooKeeper transitively; the test-jars carry
EmbeddedKafkaCluster and the embedded ZooKeeper helper. -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.13</artifactId>
<version>${kafka.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.13</artifactId>
<version>${kafka.version}</version>
<type>test-jar</type>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>${kafka.version}</version>
<type>test-jar</type>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
<type>test-jar</type>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-server-common</artifactId>
<version>${kafka.version}</version>
<type>test-jar</type>
<classifier>test</classifier>
<scope>test</scope>
</dependency>
<!-- Kafka's IntegrationTestUtils (topic-readiness waits used by
EmbeddedKafkaCluster.createTopic) asserts via Hamcrest. -->
<dependency>
<groupId>org.hamcrest</groupId>
<artifactId>hamcrest</artifactId>
<version>2.2</version>
<scope>test</scope>
</dependency>
</dependencies>

<build>
Expand Down
118 changes: 118 additions & 0 deletions kafka-streams-app/src/main/java/berlinmod/BerlinMODBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
package berlinmod;

import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;

import java.lang.reflect.Field;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.TreeMap;

/**
* Per-cell throughput benchmark for the BerlinMOD-9 × 3-form streaming matrix on
* Kafka Streams.
*
* <p>Each cell runs in isolation through its own single-cell topology
* ({@link BerlinMODTopology#buildCell}) and {@link TopologyTestDriver} over the
* corpus — the Kafka-Streams analog of MobilityFlink's per-cell jobs, so the
* per-cell wall-clock and throughput are independent and comparable. The corpus
* and the per-query parameters are corpus-derived via {@link BerlinMODCorpus};
* the spatial predicates evaluate through MEOS (see {@link MEOSBridge}).
*
* <pre>
* java … berlinmod.BerlinMODBenchmark --csv &lt;berlinmod_instants.csv&gt; [--max N]
* java … berlinmod.BerlinMODBenchmark --vehicles 50 --events 600 [--only Q3-continuous]
* </pre>
*/
public final class BerlinMODBenchmark {

private BerlinMODBenchmark() { /* utility */ }

public static void main(String[] args) throws Exception {
String csv = null, only = null;
int maxRows = 0, vehicles = 50, events = 600;
for (int i = 0; i < args.length; i++) {
switch (args[i]) {
case "--csv": csv = args[++i]; break;
case "--max": maxRows = Integer.parseInt(args[++i]); break;
case "--vehicles": vehicles = Integer.parseInt(args[++i]); break;
case "--events": events = Integer.parseInt(args[++i]); break;
case "--only": only = args[++i]; break;
default: break;
}
}
List<BerlinMODTrip> corpus = csv != null
? BerlinMODCorpus.fromInstantsCsv(csv, maxRows)
: BerlinMODCorpus.synthetic(vehicles, events);
int n = corpus.size();
long maxTs = corpus.stream().mapToLong(BerlinMODTrip::getTimestamp).max().orElse(0L);
BerlinMODCorpus.Params p = BerlinMODCorpus.derive(corpus);
System.out.printf("Corpus: %s, %d events; window=%ds tick=%dms%n",
csv != null ? "real BerlinMOD instants" : "synthetic", n, p.windowSeconds, p.snapshotTickMillis);

// (cellName, outputTopic) enumerated from the topology's *_OUTPUT constants.
TreeMap<String, String> cells = new TreeMap<>();
for (Field f : BerlinMODTopology.class.getFields()) {
if (f.getName().endsWith("_OUTPUT") && f.getType() == String.class) {
cells.put(cellName(f.getName()), (String) f.get(null));
}
}

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "berlinmod-bench");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray().getClass());
BerlinMODTripSerde tripSerde = new BerlinMODTripSerde();

List<String[]> rows = new ArrayList<>();
for (var e : cells.entrySet()) {
String cell = e.getKey(), topic = e.getValue();
if (only != null && !cell.equals(only)) {
continue;
}
long out;
long t0 = System.nanoTime();
try (TopologyTestDriver driver = new TopologyTestDriver(BerlinMODTopology.buildCell(p, cell), props)) {
TestInputTopic<Integer, BerlinMODTrip> input = driver.createInputTopic(
BerlinMODTopology.INPUT_TOPIC, new IntegerSerializer(), tripSerde.serializer());
for (BerlinMODTrip trip : corpus) {
input.pipeInput(trip.getVehicleId(), trip, Instant.ofEpochMilli(trip.getTimestamp()));
}
input.pipeInput(-1, new BerlinMODTrip(-1, maxTs + 3_600_000L, 0.0, 0.0),
Instant.ofEpochMilli(maxTs + 3_600_000L));
input.pipeInput(-1, new BerlinMODTrip(-1, maxTs + 7_200_000L, 0.0, 0.0),
Instant.ofEpochMilli(maxTs + 7_200_000L));
TestOutputTopic<byte[], byte[]> output = driver.createOutputTopic(
topic, new ByteArrayDeserializer(), new ByteArrayDeserializer());
out = output.readRecordsToList().size();
}
long wallMs = (System.nanoTime() - t0) / 1_000_000L;
double tput = wallMs > 0 ? n / (wallMs / 1000.0) : 0;
rows.add(new String[]{cell, String.valueOf(n), String.valueOf(out),
String.valueOf(wallMs), String.format("%,.0f", tput)});
System.out.printf(" %-14s out=%-8d %6d ms %,.0f ev/s%n", cell, out, wallMs, tput);
}

System.out.println();
System.out.println("| Cell | Events in | Output rows | Wall (ms) | Throughput (ev/s) |");
System.out.println("|---|---:|---:|---:|---:|");
for (String[] r : rows) {
System.out.printf("| %s | %s | %s | %s | %s |%n", r[0], r[1], r[2], r[3], r[4]);
}
}

/** Q3_CONTINUOUS_OUTPUT → Q3-continuous */
private static String cellName(String field) {
String s = field.substring(0, field.length() - "_OUTPUT".length());
int us = s.indexOf('_');
return s.substring(0, us) + "-" + s.substring(us + 1).toLowerCase();
}
}
Loading