diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4dd9b3c --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +# Maven build artefacts +target/ + +# IDE +.idea/ +.vscode/ +*.iml + +# OS +.DS_Store +Thumbs.db diff --git a/README.md b/README.md index 64ca8d3..d4cedeb 100644 --- a/README.md +++ b/README.md @@ -14,3 +14,57 @@ The MobilityDB project is developed by the Computer & Decision Engineering Depar More information about MobilityDB, including publications, presentations, etc., can be found in the MobilityDB [website](https://mobilitydb.com). + +# BerlinMOD-9 × 3 streaming forms — the parity matrix on Kafka Streams + +The streaming-side parity matrix runs all nine BerlinMOD reference queries (Q1..Q9) in three streaming forms each on this runtime: **continuous** (always-on, per-event emission), **windowed** (tumbling 10-second aggregation), and **snapshot** (5-second tick — the parity-oracle form whose output at watermark T equals the batch BerlinMOD-Q result on data up to T). + +| Q | Topic | Continuous | Windowed | Snapshot | +|---|---|---|---|---| +| Q1 | "which vehicles have appeared in the stream?" | ✓ | ✓ | ✓ | +| Q2 | "where is vehicle X at time T?" | ✓ | ✓ | ✓ | +| Q3 | "vehicles within d of P at time T?" | ✓ | ✓ | ✓ | +| Q4 | "vehicles entered region R, and when?" | ✓ | ✓ | ✓ | +| Q5 | "pairs of vehicles meeting near P" | ✓ | ✓ | ✓ | +| Q6 | "cumulative distance per vehicle" | ✓ | ✓ | ✓ | +| Q7 | "first passage of vehicles through POIs" | ✓ | ✓ | ✓ | +| Q8 | "vehicles close to a road segment" | ✓ | ✓ | ✓ | +| Q9 | "distance between vehicles X and Y at time T" | ✓ | ✓ | ✓ | + +**27 / 27 cells** = the full MobilityKafka parity-matrix row. Each cell has a dedicated `Q{Continuous,Windowed,Snapshot}Processor` class in [`kafka-streams-app/src/main/java/berlinmod/`](kafka-streams-app/src/main/java/berlinmod/) and is locally verified via [`BerlinMODQ1LocalTest`](kafka-streams-app/src/main/java/berlinmod/BerlinMODQ1LocalTest.java) running on the Kafka-Streams `TopologyTestDriver` (no real broker required). + +## Module structure + +`kafka-streams-app/` is a Maven project (Java 21, Kafka Streams 3.6.0) holding: + +- 27 per-cell `Q{Continuous,Windowed,Snapshot}Processor` classes +- `BerlinMODTopology` — unified topology fanning input topic `berlinmod` to per-Q-form output topics +- `BerlinMODTrip` + `BerlinMODTripSerde` — shared data class + JSON Serde (byte-shape equivalent to MobilityFlink's `BerlinMODTrip`) +- `Haversine` + `SegmentDistance` + `PointOfInterest` — pure-Java geometry utilities used by the spatial-predicate cells +- `BerlinMODQ1LocalTest` — TopologyTestDriver-based local end-to-end driver + +The streaming snapshot form converges to the batch BerlinMOD result on the same scale-factor corpus, anchored against the cross-platform outputs in [MobilityDB-BerlinMOD](https://github.com/MobilityDB/MobilityDB-BerlinMOD). + +Spatial predicates today use pure-Java great-circle (`Haversine`) and planar segment-distance (`SegmentDistance`) utilities; each call site is marked `TODO(meos)` for JMEOS-bridge migration after [JMEOS#15](https://github.com/MobilityDB/JMEOS/pull/15) (the MEOS 1.4 regen) settles. + +## Build and run + +``` +cd kafka-streams-app +mvn -q clean package -DskipTests +java --add-opens java.base/java.lang=ALL-UNNAMED \ + --add-opens java.base/java.util=ALL-UNNAMED \ + --add-opens java.base/java.lang.reflect=ALL-UNNAMED \ + -cp target/mobility-kafka-streams-1.0-SNAPSHOT.jar \ + berlinmod.BerlinMODQ1LocalTest +``` + +The driver pipes a 21-event sorted-event-time corpus plus two sentinel records at `t = T0+15001` and `t = T0+20001` (to step the STREAM_TIME punctuator through the desired tick boundaries) and reads every per-Q-form output topic with the appropriate deserializer. Expected per-Q-form counts are in the PR body for the open scaffold PR. + +## Sibling parity work in the ecosystem + +- [MobilityFlink#3](https://github.com/MobilityDB/MobilityFlink/pull/3) — the same 27-cell row on Flink +- [MobilityNebula#15](https://github.com/MobilityDB/MobilityNebula/pull/15) — 15 of 27 cells on NebulaStream (Q1, Q2, Q3, Q4, Q7-via-POI-fanout) +- [MobilityDB-BerlinMOD#29](https://github.com/MobilityDB/MobilityDB-BerlinMOD/pull/29) — the batch BerlinMOD-9 cross-platform timings (the snapshot form's gold-answer source) +- [MobilityDB/.github#10](https://github.com/MobilityDB/.github/pull/10) — the ecosystem-profile description of the stream-layers tier + diff --git a/kafka-streams-app/pom.xml b/kafka-streams-app/pom.xml new file mode 100644 index 0000000..4ea8465 --- /dev/null +++ b/kafka-streams-app/pom.xml @@ -0,0 +1,110 @@ + + + 4.0.0 + + com.mobilitydb.kafka + mobility-kafka-streams + 1.0-SNAPSHOT + + + 21 + 21 + UTF-8 + 3.6.0 + 2.18.2 + 2.0.16 + 5.11.4 + + + + + org.apache.kafka + kafka-streams + ${kafka.version} + + + + org.apache.kafka + kafka-streams-test-utils + ${kafka.version} + test + + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.version} + + + + org.slf4j + slf4j-api + ${slf4j.version} + + + org.slf4j + slf4j-simple + ${slf4j.version} + + + + org.junit.jupiter + junit-jupiter-api + ${junit.version} + test + + + org.junit.jupiter + junit-jupiter-engine + ${junit.version} + test + + + + + + + maven-compiler-plugin + 3.12.1 + + ${maven.compiler.source} + ${maven.compiler.target} + + + + maven-surefire-plugin + 3.2.5 + + + maven-shade-plugin + 3.5.1 + + + package + + shade + + + false + + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + + + + + + + diff --git a/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java new file mode 100644 index 0000000..3f0a88d --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTopology.java @@ -0,0 +1,293 @@ +package berlinmod; + +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Produced; +import org.apache.kafka.streams.state.StoreBuilder; +import org.apache.kafka.streams.state.Stores; + +import java.util.Arrays; +import java.util.List; + +/** + * BerlinMOD-9 unified Kafka-Streams topology. + * + *

Reads a single source topic {@link #INPUT_TOPIC} and fans out into + * per-{@code Q}-form output topics. Each output topic carries one form + * of one query. + */ +public final class BerlinMODTopology { + + public static final String INPUT_TOPIC = "berlinmod"; + public static final long SNAPSHOT_TICK_MILLIS = 5_000L; + public static final long WINDOW_SIZE_MILLIS = 10_000L; + + public static final String Q1_WINDOWED_OUTPUT = "berlinmod-q1-windowed"; + public static final String Q1_WIN_STORE = "q1-win-store"; + public static final String Q2_WINDOWED_OUTPUT = "berlinmod-q2-windowed"; + public static final String Q2_WIN_STORE = "q2-win-store"; + public static final String Q3_WINDOWED_OUTPUT = "berlinmod-q3-windowed"; + public static final String Q3_WIN_STORE = "q3-win-store"; + public static final String Q4_WINDOWED_OUTPUT = "berlinmod-q4-windowed"; + public static final String Q4_WIN_STORE = "q4-win-store"; + public static final String Q5_WINDOWED_OUTPUT = "berlinmod-q5-windowed"; + public static final String Q5_WIN_STORE = "q5-win-store"; + public static final String Q6_WINDOWED_OUTPUT = "berlinmod-q6-windowed"; + public static final String Q6_WIN_STORE = "q6-win-store"; + public static final String Q7_WINDOWED_OUTPUT = "berlinmod-q7-windowed"; + public static final String Q7_WIN_STORE = "q7-win-store"; + public static final String Q8_WINDOWED_OUTPUT = "berlinmod-q8-windowed"; + public static final String Q8_WIN_STORE = "q8-win-store"; + public static final String Q9_WINDOWED_OUTPUT = "berlinmod-q9-windowed"; + public static final String Q9_WIN_STORE = "q9-win-store"; + + // ---------- Q1 ---------- + public static final String Q1_CONTINUOUS_OUTPUT = "berlinmod-q1-continuous"; + public static final String Q1_SEEN_STORE = "q1-seen-store"; + public static final String Q1_SNAPSHOT_OUTPUT = "berlinmod-q1-snapshot"; + public static final String Q1_SNAP_STORE = "q1-snap-store"; + + // ---------- Q2 ---------- + public static final String Q2_CONTINUOUS_OUTPUT = "berlinmod-q2-continuous"; + public static final int Q2_TARGET_VEHICLE_ID = 2; + public static final String Q2_SNAPSHOT_OUTPUT = "berlinmod-q2-snapshot"; + public static final String Q2_SNAP_STORE = "q2-snap-store"; + + // ---------- Q3 ---------- + public static final String Q3_CONTINUOUS_OUTPUT = "berlinmod-q3-continuous"; + // Query params below are anchored to the canonical BerlinMOD sample's + // per-vehicle geometry (vehicles 1-5, centroids 7-18 km apart); radii use + // km-scale margins so the dwithin partition is identical under the + // pure-Java Haversine and the MEOS geodetic engines. + public static final double Q3_P_LON = 4.4322; // near vehicle 1 + public static final double Q3_P_LAT = 50.7670; + public static final double Q3_RADIUS_METRES = 5_000.0; + public static final String Q3_SNAPSHOT_OUTPUT = "berlinmod-q3-snapshot"; + public static final String Q3_SNAP_STORE = "q3-snap-store"; + + // ---------- Q4 ---------- + public static final String Q4_CONTINUOUS_OUTPUT = "berlinmod-q4-continuous"; + public static final String Q4_WAS_INSIDE_STORE = "q4-was-inside-store"; + public static final double Q4_XMIN = 4.40, Q4_YMIN = 50.74, Q4_XMAX = 4.47, Q4_YMAX = 50.86; + public static final String Q4_SNAPSHOT_OUTPUT = "berlinmod-q4-snapshot"; + public static final String Q4_SNAP_WAS_INSIDE_STORE = "q4-snap-was-inside-store"; + public static final String Q4_SNAP_ENTRIES_STORE = "q4-snap-entries-store"; + + // ---------- Q5 ---------- + public static final String Q5_CONTINUOUS_OUTPUT = "berlinmod-q5-continuous"; + public static final String Q5_LAST_POS_STORE = "q5-last-pos-store"; + public static final double Q5_P_LON = 4.3822; // midpoint of vehicles 1 and 2 + public static final double Q5_P_LAT = 50.7683; + public static final double Q5_D_P_METRES = 5_000.0; + public static final double Q5_D_MEET_METRES = 8_000.0; + public static final String Q5_SNAPSHOT_OUTPUT = "berlinmod-q5-snapshot"; + public static final String Q5_SNAP_STORE = "q5-snap-store"; + + // ---------- Q6 ---------- + public static final String Q6_CONTINUOUS_OUTPUT = "berlinmod-q6-continuous"; + public static final String Q6_STATE_STORE = "q6-state-store"; + public static final String Q6_SNAPSHOT_OUTPUT = "berlinmod-q6-snapshot"; + public static final String Q6_SNAP_STORE = "q6-snap-store"; + + // ---------- Q7 ---------- + public static final String Q7_CONTINUOUS_OUTPUT = "berlinmod-q7-continuous"; + public static final String Q7_FIRST_PASSED_STORE = "q7-first-passed-store"; + public static final List Q7_POIS = Arrays.asList( + new PointOfInterest(1, 4.3321, 50.7696, 2_000.0), // near vehicle 2 + new PointOfInterest(2, 4.4571, 50.8515, 2_000.0), // near vehicle 3 + new PointOfInterest(3, 4.4252, 50.9190, 2_000.0)); // near vehicle 5 + public static final String Q7_SNAPSHOT_OUTPUT = "berlinmod-q7-snapshot"; + public static final String Q7_SNAP_STORE = "q7-snap-store"; + + // ---------- Q8 ---------- + public static final String Q8_CONTINUOUS_OUTPUT = "berlinmod-q8-continuous"; + public static final double Q8_S1_LON = 4.3321, Q8_S1_LAT = 50.7696; // vehicle 2 + public static final double Q8_S2_LON = 4.3063, Q8_S2_LAT = 50.8825; // vehicle 4 + public static final double Q8_RADIUS_METRES = 5_000.0; + public static final String Q8_SNAPSHOT_OUTPUT = "berlinmod-q8-snapshot"; + public static final String Q8_SNAP_STORE = "q8-snap-store"; + + // ---------- Q9 ---------- + public static final String Q9_CONTINUOUS_OUTPUT = "berlinmod-q9-continuous"; + public static final String Q9_STATE_STORE = "q9-state-store"; + public static final int Q9_X_VEHICLE_ID = 1; + public static final int Q9_Y_VEHICLE_ID = 2; + public static final String Q9_SNAPSHOT_OUTPUT = "berlinmod-q9-snapshot"; + public static final String Q9_SNAP_STORE = "q9-snap-store"; + + private BerlinMODTopology() {} + + public static Topology build() { + StreamsBuilder builder = new StreamsBuilder(); + BerlinMODTripSerde tripSerde = new BerlinMODTripSerde(); + + // ---- continuous-form state stores ---- + addStore(builder, Q1_SEEN_STORE, Serdes.Integer(), Serdes.Boolean()); + addStore(builder, Q4_WAS_INSIDE_STORE, Serdes.Integer(), Serdes.Boolean()); + addStore(builder, Q5_LAST_POS_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q6_STATE_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q7_FIRST_PASSED_STORE, Serdes.Integer(), Serdes.Long()); + addStore(builder, Q9_STATE_STORE, Serdes.Integer(), Serdes.String()); + + // ---- windowed-form state stores ---- + addStore(builder, Q1_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q2_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q3_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q4_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q5_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q6_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q7_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q8_WIN_STORE, Serdes.Long(), Serdes.String()); + addStore(builder, Q9_WIN_STORE, Serdes.Long(), Serdes.String()); + + // ---- snapshot-form state stores (separate to avoid co-write conflicts with continuous) ---- + addStore(builder, Q1_SNAP_STORE, Serdes.Integer(), Serdes.Long()); + addStore(builder, Q2_SNAP_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q3_SNAP_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q4_SNAP_WAS_INSIDE_STORE, Serdes.Integer(), Serdes.Boolean()); + addStore(builder, Q4_SNAP_ENTRIES_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q5_SNAP_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q6_SNAP_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q7_SNAP_STORE, Serdes.Integer(), Serdes.Long()); + addStore(builder, Q8_SNAP_STORE, Serdes.Integer(), Serdes.String()); + addStore(builder, Q9_SNAP_STORE, Serdes.Integer(), Serdes.String()); + + // ---- streams ---- + KStream trips = + builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), tripSerde)); + + // Re-keyed by constant for the shared-state snapshot/multi-vehicle processors + KStream tripsK0 = trips.selectKey((k, v) -> 0); + + // ====== continuous form ====== + trips.process(() -> new Q1ContinuousProcessor(Q1_SEEN_STORE), Q1_SEEN_STORE) + .to(Q1_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); + + trips.process(() -> new Q2ContinuousProcessor(Q2_TARGET_VEHICLE_ID)) + .to(Q2_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), tripSerde)); + + trips.process(() -> new Q3ContinuousProcessor(Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES)) + .to(Q3_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Boolean())); + + trips.process(() -> new Q4ContinuousProcessor(Q4_WAS_INSIDE_STORE, Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX), + Q4_WAS_INSIDE_STORE) + .to(Q4_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); + + tripsK0.process(() -> new Q5ContinuousProcessor(Q5_LAST_POS_STORE, + Q5_P_LON, Q5_P_LAT, Q5_D_P_METRES, Q5_D_MEET_METRES), + Q5_LAST_POS_STORE) + .to(Q5_CONTINUOUS_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); + + trips.process(() -> new Q6ContinuousProcessor(Q6_STATE_STORE), Q6_STATE_STORE) + .to(Q6_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Double())); + + trips.process(() -> new Q7ContinuousProcessor(Q7_FIRST_PASSED_STORE, Q7_POIS), + Q7_FIRST_PASSED_STORE) + .to(Q7_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Long())); + + trips.process(() -> new Q8ContinuousProcessor(Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, Q8_RADIUS_METRES)) + .to(Q8_CONTINUOUS_OUTPUT, Produced.with(Serdes.Integer(), Serdes.Boolean())); + + tripsK0.process(() -> new Q9ContinuousProcessor(Q9_STATE_STORE, Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID), + Q9_STATE_STORE) + .to(Q9_CONTINUOUS_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + + // ====== windowed form (distinct-count per tumbling window for Q1/Q3/Q8) ====== + tripsK0.process(() -> new Q1WindowedProcessor(Q1_WIN_STORE, WINDOW_SIZE_MILLIS), Q1_WIN_STORE) + .to(Q1_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); + + tripsK0.process(() -> new Q3WindowedProcessor(Q3_WIN_STORE, + Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES, + WINDOW_SIZE_MILLIS), + Q3_WIN_STORE) + .to(Q3_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); + + tripsK0.process(() -> new Q8WindowedProcessor(Q8_WIN_STORE, + Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, + Q8_RADIUS_METRES, WINDOW_SIZE_MILLIS), + Q8_WIN_STORE) + .to(Q8_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Long())); + + tripsK0.process(() -> new Q2WindowedProcessor(Q2_WIN_STORE, Q2_TARGET_VEHICLE_ID, WINDOW_SIZE_MILLIS), + Q2_WIN_STORE) + .to(Q2_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q4WindowedProcessor(Q4_WIN_STORE, + Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX, + WINDOW_SIZE_MILLIS), + Q4_WIN_STORE) + .to(Q4_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q5WindowedProcessor(Q5_WIN_STORE, + Q5_P_LON, Q5_P_LAT, Q5_D_P_METRES, Q5_D_MEET_METRES, + WINDOW_SIZE_MILLIS), + Q5_WIN_STORE) + .to(Q5_WINDOWED_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); + + tripsK0.process(() -> new Q6WindowedProcessor(Q6_WIN_STORE, WINDOW_SIZE_MILLIS), Q6_WIN_STORE) + .to(Q6_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q7WindowedProcessor(Q7_WIN_STORE, Q7_POIS, WINDOW_SIZE_MILLIS), Q7_WIN_STORE) + .to(Q7_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q9WindowedProcessor(Q9_WIN_STORE, + Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID, + WINDOW_SIZE_MILLIS), + Q9_WIN_STORE) + .to(Q9_WINDOWED_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + + // ====== snapshot form (all via constant key, with STREAM_TIME punctuators) ====== + tripsK0.process(() -> new Q1SnapshotProcessor(Q1_SNAP_STORE, SNAPSHOT_TICK_MILLIS), Q1_SNAP_STORE) + .to(Q1_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); + + tripsK0.process(() -> new Q2SnapshotProcessor(Q2_SNAP_STORE, Q2_TARGET_VEHICLE_ID, SNAPSHOT_TICK_MILLIS), + Q2_SNAP_STORE) + .to(Q2_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q3SnapshotProcessor(Q3_SNAP_STORE, + Q3_P_LON, Q3_P_LAT, Q3_RADIUS_METRES, SNAPSHOT_TICK_MILLIS), + Q3_SNAP_STORE) + .to(Q3_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); + + tripsK0.process(() -> new Q4SnapshotProcessor(Q4_SNAP_WAS_INSIDE_STORE, Q4_SNAP_ENTRIES_STORE, + Q4_XMIN, Q4_YMIN, Q4_XMAX, Q4_YMAX, SNAPSHOT_TICK_MILLIS), + Q4_SNAP_WAS_INSIDE_STORE, Q4_SNAP_ENTRIES_STORE) + .to(Q4_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q5SnapshotProcessor(Q5_SNAP_STORE, + Q5_P_LON, Q5_P_LAT, Q5_D_P_METRES, Q5_D_MEET_METRES, + SNAPSHOT_TICK_MILLIS), + Q5_SNAP_STORE) + .to(Q5_SNAPSHOT_OUTPUT, Produced.with(Serdes.String(), Serdes.Double())); + + tripsK0.process(() -> new Q6SnapshotProcessor(Q6_SNAP_STORE, SNAPSHOT_TICK_MILLIS), Q6_SNAP_STORE) + .to(Q6_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q7SnapshotProcessor(Q7_SNAP_STORE, Q7_POIS, SNAPSHOT_TICK_MILLIS), Q7_SNAP_STORE) + .to(Q7_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.String())); + + tripsK0.process(() -> new Q8SnapshotProcessor(Q8_SNAP_STORE, + Q8_S1_LON, Q8_S1_LAT, Q8_S2_LON, Q8_S2_LAT, + Q8_RADIUS_METRES, SNAPSHOT_TICK_MILLIS), + Q8_SNAP_STORE) + .to(Q8_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Integer())); + + tripsK0.process(() -> new Q9SnapshotProcessor(Q9_SNAP_STORE, + Q9_X_VEHICLE_ID, Q9_Y_VEHICLE_ID, SNAPSHOT_TICK_MILLIS), + Q9_SNAP_STORE) + .to(Q9_SNAPSHOT_OUTPUT, Produced.with(Serdes.Long(), Serdes.Double())); + + return builder.build(); + } + + private static void addStore(StreamsBuilder builder, String name, + org.apache.kafka.common.serialization.Serde ks, + org.apache.kafka.common.serialization.Serde vs) { + StoreBuilder> sb = + Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(name), ks, vs); + builder.addStateStore(sb); + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/BerlinMODTrip.java b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTrip.java new file mode 100644 index 0000000..f512fe7 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTrip.java @@ -0,0 +1,37 @@ +package berlinmod; + +import java.io.Serializable; + +/** + * Plain data class for a single GPS event from a BerlinMOD trip. + * + *

Same shape as the MobilityFlink {@code BerlinMODTrip} so a CSV produced + * by the BerlinMOD generator at any SF feeds both pipelines unchanged. + */ +public class BerlinMODTrip implements Serializable { + + private static final long serialVersionUID = 1L; + + private long timestamp; // epoch milliseconds (event time) + private int vehicleId; + private double lon; + private double lat; + + public BerlinMODTrip() {} + + public BerlinMODTrip(int vehicleId, long timestamp, double lon, double lat) { + this.vehicleId = vehicleId; + this.timestamp = timestamp; + this.lon = lon; + this.lat = lat; + } + + public long getTimestamp() { return timestamp; } + public void setTimestamp(long timestamp) { this.timestamp = timestamp; } + public int getVehicleId() { return vehicleId; } + public void setVehicleId(int vehicleId) { this.vehicleId = vehicleId; } + public double getLon() { return lon; } + public void setLon(double lon) { this.lon = lon; } + public double getLat() { return lat; } + public void setLat(double lat) { this.lat = lat; } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/BerlinMODTripSerde.java b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTripSerde.java new file mode 100644 index 0000000..298fd07 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/BerlinMODTripSerde.java @@ -0,0 +1,65 @@ +package berlinmod; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serializer; + +/** + * JSON-based Kafka Streams {@link Serde} for {@link BerlinMODTrip}. + * + *

Records on the {@code berlinmod} topic are JSON objects of the form + * {@code {"t": "yyyy-MM-dd HH:mm:ss", "vehicle_id": 42, "lon": 4.36, "lat": + * 50.84}}. The serializer writes that exact shape; the deserializer parses + * the same shape (with a tolerant timestamp format matching the producer). + */ +public final class BerlinMODTripSerde implements Serde { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @Override + public Serializer serializer() { + return (topic, trip) -> { + if (trip == null) return null; + try { + return MAPPER.writeValueAsBytes(new SerForm(trip)); + } catch (Exception e) { + throw new RuntimeException("BerlinMODTrip serialize failed", e); + } + }; + } + + @Override + public Deserializer deserializer() { + return (topic, bytes) -> { + if (bytes == null) return null; + try { + SerForm sf = MAPPER.readValue(bytes, SerForm.class); + return sf.toTrip(); + } catch (Exception e) { + throw new RuntimeException("BerlinMODTrip deserialize failed", e); + } + }; + } + + /** Wire form used for JSON serialisation — keeps Jackson happy without per-field annotations. */ + private static final class SerForm { + public Long t; + public int vehicle_id; + public double lon; + public double lat; + + public SerForm() {} + + SerForm(BerlinMODTrip trip) { + this.t = trip.getTimestamp(); + this.vehicle_id = trip.getVehicleId(); + this.lon = trip.getLon(); + this.lat = trip.getLat(); + } + + BerlinMODTrip toTrip() { + return new BerlinMODTrip(vehicle_id, t == null ? 0L : t, lon, lat); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Haversine.java b/kafka-streams-app/src/main/java/berlinmod/Haversine.java new file mode 100644 index 0000000..4cd723e --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Haversine.java @@ -0,0 +1,35 @@ +package berlinmod; + +/** + * Great-circle distance in metres between two WGS84 (lon, lat) points. + * + *

Matches the MobilityFlink {@code Haversine} utility verbatim so the + * spatial-predicate cells (Q3 in this PR; Q4–Q9 in follow-ups) produce + * cross-platform byte-identical output against the same synthetic corpus. + * TODO(meos): cells using this utility migrate to the MEOS equivalent + * (edwithin_tgeo_geo / NAD / etc.) via JMEOS once the in-flight 1.4 bump + * signals settled. + */ +public final class Haversine { + + private static final double EARTH_RADIUS_METRES = 6_371_000.0; + + private Haversine() {} + + public static double distanceMetres(double lon1, double lat1, double lon2, double lat2) { + double phi1 = Math.toRadians(lat1); + double phi2 = Math.toRadians(lat2); + double dPhi = Math.toRadians(lat2 - lat1); + double dLambda = Math.toRadians(lon2 - lon1); + + double a = Math.sin(dPhi / 2) * Math.sin(dPhi / 2) + + Math.cos(phi1) * Math.cos(phi2) + * Math.sin(dLambda / 2) * Math.sin(dLambda / 2); + double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a)); + return EARTH_RADIUS_METRES * c; + } + + public static boolean withinMetres(double lon, double lat, double pLon, double pLat, double radiusMetres) { + return distanceMetres(lon, lat, pLon, pLat) <= radiusMetres; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/PointOfInterest.java b/kafka-streams-app/src/main/java/berlinmod/PointOfInterest.java new file mode 100644 index 0000000..067f804 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/PointOfInterest.java @@ -0,0 +1,23 @@ +package berlinmod; + +import java.io.Serializable; + +/** + * Point-of-interest record used by BerlinMOD-Q7: an integer id, a (lon, lat) + * location, and a proximity radius in metres. + */ +public final class PointOfInterest implements Serializable { + private static final long serialVersionUID = 1L; + + public final int id; + public final double lon; + public final double lat; + public final double radiusMetres; + + public PointOfInterest(int id, double lon, double lat, double radiusMetres) { + this.id = id; + this.lon = lon; + this.lat = lat; + this.radiusMetres = radiusMetres; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q1ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q1ContinuousProcessor.java new file mode 100644 index 0000000..08ce438 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q1ContinuousProcessor.java @@ -0,0 +1,56 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * BerlinMOD-Q1 — continuous form, Kafka-Streams Processor API. + * + *

"Which vehicles have appeared in the stream?" + * + *

Uses a {@link KeyValueStore} keyed by {@code vehicleId} (named in the + * caller) to dedupe — emits {@code (vehicleId, firstSeenTimestamp)} the + * first time each vehicle is seen and ignores subsequent events. + * + *

Same semantic as the MobilityFlink {@code Q1ContinuousFunction}; the + * differences are purely in the runtime API (Kafka Streams Processor vs + * Flink {@code KeyedProcessFunction}). + */ +public class Q1ContinuousProcessor implements Processor { + + private static final Logger LOG = LoggerFactory.getLogger(Q1ContinuousProcessor.class); + + private final String storeName; + private KeyValueStore seen; + private ProcessorContext ctx; + + public Q1ContinuousProcessor(String storeName) { + this.storeName = storeName; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.seen = context.getStateStore(storeName); + } + + @Override + public void process(Record record) { + Integer vehicleId = record.key(); + BerlinMODTrip trip = record.value(); + if (vehicleId == null || trip == null || vehicleId == -1) return; + + Boolean alreadySeen = seen.get(vehicleId); + if (alreadySeen == null || !alreadySeen) { + seen.put(vehicleId, true); + ctx.forward(new Record<>(vehicleId, trip.getTimestamp(), trip.getTimestamp())); + if (LOG.isDebugEnabled()) { + LOG.debug("Q1-continuous first-sighting: vehicle={} t={}", vehicleId, trip.getTimestamp()); + } + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q1SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q1SnapshotProcessor.java new file mode 100644 index 0000000..56ed8ea --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q1SnapshotProcessor.java @@ -0,0 +1,74 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q1 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, which vehicles have appeared in the stream up to T?" + * + *

The parity-oracle form: streaming output at watermark T equals the + * batch BerlinMOD-Q1 result on data up to T. + * + *

Caller keys the input by a constant so the shared cross-vehicle "seen" + * map lives in a single subtask. On each event (sentinel vehicleId == -1 + * is ignored), record vehicleId → firstSeenTime if not already present. + * On each STREAM_TIME punctuator fire (every {@code snapshotTickMillis}), + * walk the map and forward {@code (currentStreamTime, vehicleId)} per + * recorded vehicle, sorted by vehicleId for deterministic output. + */ +public class Q1SnapshotProcessor implements Processor { + + private final String storeName; + private final long snapshotTickMillis; + private KeyValueStore seen; + private ProcessorContext ctx; + + public Q1SnapshotProcessor(String storeName, long snapshotTickMillis) { + this.storeName = storeName; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.seen = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + if (seen.get(trip.getVehicleId()) == null) { + seen.put(trip.getVehicleId(), trip.getTimestamp()); + } + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List ids = new ArrayList<>(); + try (KeyValueIterator it = seen.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + ids.add(kv.key); + } + } + ids.sort(Comparator.naturalOrder()); + for (Integer vehicleId : ids) { + ctx.forward(new Record<>(tick, vehicleId, tick)); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q1WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q1WindowedProcessor.java new file mode 100644 index 0000000..ea12251 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q1WindowedProcessor.java @@ -0,0 +1,95 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * BerlinMOD-Q1 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, how many distinct vehicles appeared + * in the window?" + * + *

State value is a comma-separated set of vehicleIds seen in each + * window start. STREAM_TIME punctuator at {@code windowSizeMs} interval + * emits closed windows and removes them from the store. + */ +public class Q1WindowedProcessor implements Processor { + + private final String storeName; + private final long windowSizeMs; + private KeyValueStore winState; // winStart -> "vid1,vid2,..." + private ProcessorContext ctx; + + public Q1WindowedProcessor(String storeName, long windowSizeMs) { + this.storeName = storeName; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String prior = winState.get(winStart); + String sv = Integer.toString(trip.getVehicleId()); + if (prior == null) { + winState.put(winStart, sv); + } else if (!containsId(prior, sv)) { + winState.put(winStart, prior + "," + sv); + } + } + + private void punctuate(long currentStreamTime) { + // Emit closed windows (winEnd <= currentStreamTime) and remove + List toEmit = new ArrayList<>(); + List counts = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + toEmit.add(kv.key); + Set distinct = new HashSet<>(); + for (String s : kv.value.split(",")) distinct.add(s); + counts.add((long) distinct.size()); + } + } + } + // Sort by winStart for deterministic order + Integer[] idx = new Integer[toEmit.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(toEmit::get)); + for (Integer i : idx) { + long winStart = toEmit.get(i); + ctx.forward(new Record<>(winStart, counts.get(i), winStart + windowSizeMs - 1)); + winState.delete(winStart); + } + } + + private boolean containsId(String csv, String id) { + if (csv == null) return false; + for (String s : csv.split(",")) { + if (s.equals(id)) return true; + } + return false; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q2ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q2ContinuousProcessor.java new file mode 100644 index 0000000..e5881af --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q2ContinuousProcessor.java @@ -0,0 +1,36 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; + +/** + * BerlinMOD-Q2 — continuous form, Kafka-Streams Processor API. + * + *

"Where is vehicle X right now?" + * + *

Pure stateless filter: forward records whose key matches the queried + * {@code targetVehicleId}, drop the rest. Matches the MobilityFlink + * {@code Q2ContinuousFunction} pattern. + */ +public class Q2ContinuousProcessor implements Processor { + + private final int targetVehicleId; + private ProcessorContext ctx; + + public Q2ContinuousProcessor(int targetVehicleId) { + this.targetVehicleId = targetVehicleId; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + } + + @Override + public void process(Record record) { + if (record.key() != null && record.key() == targetVehicleId) { + ctx.forward(record); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q2SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q2SnapshotProcessor.java new file mode 100644 index 0000000..bdca33d --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q2SnapshotProcessor.java @@ -0,0 +1,60 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; + +/** + * BerlinMOD-Q2 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, where is vehicle X?" using X's most-recent-known + * position on or before T. + * + *

Single-key state (key=0) value "lon,lat,t" updated only when an event + * arrives for the queried {@code targetVehicleId}. STREAM_TIME punctuator + * every {@code snapshotTickMillis} emits {@code (currentTick, lon, lat, + * lastEventT)} when state is set. + */ +public class Q2SnapshotProcessor implements Processor { + + private final String storeName; + private final int targetVehicleId; + private final long snapshotTickMillis; + private KeyValueStore state; + private ProcessorContext ctx; + + public Q2SnapshotProcessor(String storeName, int targetVehicleId, long snapshotTickMillis) { + this.storeName = storeName; + this.targetVehicleId = targetVehicleId; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.state = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + if (trip.getVehicleId() == targetVehicleId) { + state.put(0, trip.getLon() + "," + trip.getLat() + "," + trip.getTimestamp()); + } + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + String v = state.get(0); + if (v != null) { + ctx.forward(new Record<>(tick, v, tick)); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q2WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q2WindowedProcessor.java new file mode 100644 index 0000000..c5fb102 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q2WindowedProcessor.java @@ -0,0 +1,78 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q2 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, last-known (lon, lat) of vehicle X + * seen in the window." + * + *

State value "lon,lat,t" overwritten on every X event whose timestamp + * falls in {@code winStart}. STREAM_TIME punctuator at {@code windowSizeMs} + * emits closed windows. + */ +public class Q2WindowedProcessor implements Processor { + + private final String storeName; + private final int targetVehicleId; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q2WindowedProcessor(String storeName, int targetVehicleId, long windowSizeMs) { + this.storeName = storeName; + this.targetVehicleId = targetVehicleId; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + if (trip.getVehicleId() != targetVehicleId) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + winState.put(winStart, trip.getLon() + "," + trip.getLat() + "," + trip.getTimestamp()); + } + + private void punctuate(long currentStreamTime) { + List toEmit = new ArrayList<>(); + List values = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + toEmit.add(kv.key); + values.add(kv.value); + } + } + } + Integer[] idx = new Integer[toEmit.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(toEmit::get)); + for (Integer i : idx) { + long winStart = toEmit.get(i); + ctx.forward(new Record<>(winStart, values.get(i), winStart + windowSizeMs - 1)); + winState.delete(winStart); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q3ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q3ContinuousProcessor.java new file mode 100644 index 0000000..ca91889 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q3ContinuousProcessor.java @@ -0,0 +1,46 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; + +/** + * BerlinMOD-Q3 — continuous form, Kafka-Streams Processor API. + * + *

"Is this vehicle currently within {@code d} metres of point P?" + * + *

Stateless per-event predicate: forward {@code (vehicleId, + * eventTime, near)} per incoming GPS event. Same predicate semantics + * as MobilityFlink's {@code Q3ContinuousFunction}. + * + *

Predicate today: pure-Java great-circle distance (see {@link Haversine}). + * TODO(meos): replace with the MEOS {@code edwithin_tgeo_geo} operator via + * the JMEOS bridge. + */ +public class Q3ContinuousProcessor implements Processor { + + private final double pLon; + private final double pLat; + private final double radiusMetres; + private ProcessorContext ctx; + + public Q3ContinuousProcessor(double pLon, double pLat, double radiusMetres) { + this.pLon = pLon; + this.pLat = pLat; + this.radiusMetres = radiusMetres; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + boolean near = Haversine.withinMetres( + trip.getLon(), trip.getLat(), pLon, pLat, radiusMetres); + ctx.forward(new Record<>(trip.getVehicleId(), near, trip.getTimestamp())); + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q3SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q3SnapshotProcessor.java new file mode 100644 index 0000000..b9c6063 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q3SnapshotProcessor.java @@ -0,0 +1,79 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q3 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, which vehicles are within {@code d} metres of point P?" + * using each vehicle's most-recent-known position on or before T. + * + *

Caller keys the input by a constant so the shared cross-vehicle + * last-known store lives in one subtask. Per event: update last-known. + * Per STREAM_TIME punctuator fire: iterate last-known, evaluate the + * Haversine radius predicate, forward {@code (currentTick, vehicleId)} + * for every near vehicle (sorted by vehicleId). + */ +public class Q3SnapshotProcessor implements Processor { + + private final String storeName; + private final double pLon, pLat, radiusMetres; + private final long snapshotTickMillis; + private KeyValueStore lastPos; // vehicleId -> "lon,lat" + private ProcessorContext ctx; + + public Q3SnapshotProcessor(String storeName, double pLon, double pLat, + double radiusMetres, long snapshotTickMillis) { + this.storeName = storeName; + this.pLon = pLon; + this.pLat = pLat; + this.radiusMetres = radiusMetres; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.lastPos = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + lastPos.put(trip.getVehicleId(), trip.getLon() + "," + trip.getLat()); + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List nearIds = new ArrayList<>(); + try (KeyValueIterator it = lastPos.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + String[] ll = kv.value.split(",", 2); + double lon = Double.parseDouble(ll[0]); + double lat = Double.parseDouble(ll[1]); + if (Haversine.withinMetres(lon, lat, pLon, pLat, radiusMetres)) { + nearIds.add(kv.key); + } + } + } + nearIds.sort(Comparator.naturalOrder()); + for (Integer vid : nearIds) { + ctx.forward(new Record<>(tick, vid, tick)); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q3WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q3WindowedProcessor.java new file mode 100644 index 0000000..0803aac --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q3WindowedProcessor.java @@ -0,0 +1,97 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * BerlinMOD-Q3 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, how many distinct vehicles were + * within {@code d} metres of point P at any time during the window?" + * + *

Same shape as {@link Q1WindowedProcessor} but only records vehicles + * whose event satisfies the radius predicate. + */ +public class Q3WindowedProcessor implements Processor { + + private final String storeName; + private final double pLon, pLat, radiusMetres; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q3WindowedProcessor(String storeName, double pLon, double pLat, + double radiusMetres, long windowSizeMs) { + this.storeName = storeName; + this.pLon = pLon; + this.pLat = pLat; + this.radiusMetres = radiusMetres; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + if (!Haversine.withinMetres(trip.getLon(), trip.getLat(), pLon, pLat, radiusMetres)) { + return; + } + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String prior = winState.get(winStart); + String sv = Integer.toString(trip.getVehicleId()); + if (prior == null) { + winState.put(winStart, sv); + } else if (!Q1WindowedProcessor.class.getName().isEmpty() && !contains(prior, sv)) { + winState.put(winStart, prior + "," + sv); + } + } + + private void punctuate(long currentStreamTime) { + List toEmit = new ArrayList<>(); + List counts = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + toEmit.add(kv.key); + Set distinct = new HashSet<>(); + for (String s : kv.value.split(",")) distinct.add(s); + counts.add((long) distinct.size()); + } + } + } + Integer[] idx = new Integer[toEmit.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(toEmit::get)); + for (Integer i : idx) { + long winStart = toEmit.get(i); + ctx.forward(new Record<>(winStart, counts.get(i), winStart + windowSizeMs - 1)); + winState.delete(winStart); + } + } + + private static boolean contains(String csv, String id) { + for (String s : csv.split(",")) if (s.equals(id)) return true; + return false; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q4ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q4ContinuousProcessor.java new file mode 100644 index 0000000..c386b89 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q4ContinuousProcessor.java @@ -0,0 +1,57 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; + +/** + * BerlinMOD-Q4 — continuous form, Kafka-Streams Processor API. + * + *

"Which vehicles entered region R (transition outside → inside)?" + * + *

Keyed by vehicleId. Per-vehicle state tracks the last seen + * inside-or-outside flag for R; on each event, detect outside → inside + * transition and emit {@code (vehicleId, entryTime)}. + * + *

TODO(meos): replace the point-in-box predicate with a MEOS + * {@code eintersects} call against an STBox via the JMEOS bridge. + */ +public class Q4ContinuousProcessor implements Processor { + + private final String storeName; + private final double xmin, ymin, xmax, ymax; + private KeyValueStore wasInside; + private ProcessorContext ctx; + + public Q4ContinuousProcessor(String storeName, double xmin, double ymin, double xmax, double ymax) { + this.storeName = storeName; + this.xmin = xmin; + this.ymin = ymin; + this.xmax = xmax; + this.ymax = ymax; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.wasInside = context.getStateStore(storeName); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + boolean isInside = inBox(trip.getLon(), trip.getLat()); + Boolean prev = wasInside.get(trip.getVehicleId()); + boolean prevInside = prev != null && prev; + if (isInside && !prevInside) { + ctx.forward(new Record<>(trip.getVehicleId(), trip.getTimestamp(), trip.getTimestamp())); + } + wasInside.put(trip.getVehicleId(), isInside); + } + + private boolean inBox(double lon, double lat) { + return lon >= xmin && lon <= xmax && lat >= ymin && lat <= ymax; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q4SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q4SnapshotProcessor.java new file mode 100644 index 0000000..27a1383 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q4SnapshotProcessor.java @@ -0,0 +1,99 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q4 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, what is the list of (vehicleId, entryTime) pairs for all + * vehicles that entered region R at or before T?" + * + *

Caller keys the input by a constant so the shared cross-vehicle state + * lives in one subtask. Per-vehicle two stores: {@code wasInside} (Boolean) + * and {@code entries} (semicolon-separated entry times). Per event: detect + * outside→inside transition and append the entry time. Per STREAM_TIME + * punctuator fire: walk every vehicle, emit one + * {@code (currentTick, vehicleId, entryTime)} per recorded entry time + * ≤ currentTick. + */ +public class Q4SnapshotProcessor implements Processor { + + private final String wasInsideStoreName; + private final String entriesStoreName; + private final double xmin, ymin, xmax, ymax; + private final long snapshotTickMillis; + private KeyValueStore wasInside; + private KeyValueStore entries; // vehicleId -> "t1;t2;..." + private ProcessorContext ctx; + + public Q4SnapshotProcessor(String wasInsideStoreName, String entriesStoreName, + double xmin, double ymin, double xmax, double ymax, + long snapshotTickMillis) { + this.wasInsideStoreName = wasInsideStoreName; + this.entriesStoreName = entriesStoreName; + this.xmin = xmin; + this.ymin = ymin; + this.xmax = xmax; + this.ymax = ymax; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.wasInside = context.getStateStore(wasInsideStoreName); + this.entries = context.getStateStore(entriesStoreName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + boolean curr = trip.getLon() >= xmin && trip.getLon() <= xmax + && trip.getLat() >= ymin && trip.getLat() <= ymax; + Boolean prev = wasInside.get(trip.getVehicleId()); + boolean prevInside = prev != null && prev; + if (curr && !prevInside) { + String prior = entries.get(trip.getVehicleId()); + String updated = (prior == null || prior.isEmpty()) + ? Long.toString(trip.getTimestamp()) + : prior + ";" + trip.getTimestamp(); + entries.put(trip.getVehicleId(), updated); + } + wasInside.put(trip.getVehicleId(), curr); + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List ids = new ArrayList<>(); + try (KeyValueIterator it = entries.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.value != null && !kv.value.isEmpty()) ids.add(kv.key); + } + } + ids.sort(Comparator.naturalOrder()); + for (Integer vid : ids) { + String list = entries.get(vid); + for (String s : list.split(";")) { + long entryTime = Long.parseLong(s); + if (entryTime <= tick) { + ctx.forward(new Record<>(tick, vid + "@" + entryTime, tick)); + } + } + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q4WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q4WindowedProcessor.java new file mode 100644 index 0000000..346ec34 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q4WindowedProcessor.java @@ -0,0 +1,143 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q4 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, which vehicles entered region R during + * the window?" Intra-window scoping: a vehicle's first event in the + * window with {@code inBox(...) == true} counts as an entry (no cross-window + * memory of prior inside-state). + * + *

State value encodes the entries already recorded for the window plus + * a "last seen inside" flag per vehicle: comma-separated list of + * {@code "vid:wasInside:entryTime"} triples. Tracks per-(window, vehicle) + * to detect intra-window outside→inside transitions. + */ +public class Q4WindowedProcessor implements Processor { + + private final String storeName; + private final double xmin, ymin, xmax, ymax; + private final long windowSizeMs; + private KeyValueStore winState; // winStart -> "vid:wasInside:entries|..." + private ProcessorContext ctx; + + public Q4WindowedProcessor(String storeName, double xmin, double ymin, double xmax, double ymax, + long windowSizeMs) { + this.storeName = storeName; + this.xmin = xmin; + this.ymin = ymin; + this.xmax = xmax; + this.ymax = ymax; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + boolean curr = trip.getLon() >= xmin && trip.getLon() <= xmax + && trip.getLat() >= ymin && trip.getLat() <= ymax; + String s = winState.get(winStart); + // Parse per-vehicle records separated by '|' + StringBuilder rebuilt = new StringBuilder(); + boolean foundVehicle = false; + if (s != null && !s.isEmpty()) { + for (String chunk : s.split("\\|")) { + String[] f = chunk.split(":", 3); + int vid = Integer.parseInt(f[0]); + boolean wasInside = Boolean.parseBoolean(f[1]); + String entries = f.length > 2 ? f[2] : ""; + if (vid == trip.getVehicleId()) { + foundVehicle = true; + String newEntries = entries; + if (curr && !wasInside) { + newEntries = entries.isEmpty() + ? Long.toString(trip.getTimestamp()) + : entries + "," + trip.getTimestamp(); + } + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(vid).append(":").append(curr).append(":").append(newEntries); + } else { + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(chunk); + } + } + } + if (!foundVehicle) { + // First event for this vehicle in this window — intra-window scoping + // treats first-seen-inside as an entry. + String entries = curr ? Long.toString(trip.getTimestamp()) : ""; + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(trip.getVehicleId()).append(":").append(curr).append(":").append(entries); + } + winState.put(winStart, rebuilt.toString()); + } + + private void punctuate(long currentStreamTime) { + List closedStarts = new ArrayList<>(); + List closedStates = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + closedStarts.add(kv.key); + closedStates.add(kv.value); + } + } + } + Integer[] idx = new Integer[closedStarts.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(closedStarts::get)); + for (Integer i : idx) { + long winStart = closedStarts.get(i); + // Emit one "vid:entryTime" per recorded entry, sorted by (vid, entryTime) + List vidIdx = new ArrayList<>(); + List times = new ArrayList<>(); + for (String chunk : closedStates.get(i).split("\\|")) { + if (chunk.isEmpty()) continue; + String[] f = chunk.split(":", 3); + int vid = Integer.parseInt(f[0]); + String entries = f.length > 2 ? f[2] : ""; + if (entries.isEmpty()) continue; + for (String s : entries.split(",")) { + vidIdx.add(new int[]{vid}); + times.add(new long[]{Long.parseLong(s)}); + } + } + // Stable sort by vid then time + Integer[] sortedIdx = new Integer[vidIdx.size()]; + for (int k = 0; k < sortedIdx.length; k++) sortedIdx[k] = k; + java.util.Arrays.sort(sortedIdx, Comparator + .comparingInt((Integer k) -> vidIdx.get(k)[0]) + .thenComparingLong(k -> times.get(k)[0])); + for (Integer k : sortedIdx) { + ctx.forward(new Record<>(winStart, + vidIdx.get(k)[0] + ":" + times.get(k)[0], + winStart + windowSizeMs - 1)); + } + winState.delete(winStart); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q5ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q5ContinuousProcessor.java new file mode 100644 index 0000000..804a6eb --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q5ContinuousProcessor.java @@ -0,0 +1,94 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q5 — continuous form, Kafka-Streams Processor API. + * + *

"Which pairs of vehicles are currently meeting near point P?" + * + *

Caller should key the input stream by a constant so the shared + * cross-vehicle last-known state lives in one subtask. Per-event: + * update last-known of the event's vehicle, then enumerate all known + * pairs and forward {@code (a, b, eventTime, distanceMetres)} encoded + * as a string key + Double value for every currently-meeting pair + * (a < b for stable identity). + * + *

State value encoded as "lon,lat" string; the encoding is private + * to this processor (avoids declaring a tuple SerDe for the scaffold). + */ +public class Q5ContinuousProcessor implements Processor { + + private final String storeName; + private final double pLon, pLat, dPMetres, dMeetMetres; + private KeyValueStore lastPos; + private ProcessorContext ctx; + + public Q5ContinuousProcessor(String storeName, double pLon, double pLat, + double dPMetres, double dMeetMetres) { + this.storeName = storeName; + this.pLon = pLon; + this.pLat = pLat; + this.dPMetres = dPMetres; + this.dMeetMetres = dMeetMetres; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.lastPos = context.getStateStore(storeName); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + lastPos.put(trip.getVehicleId(), trip.getLon() + "," + trip.getLat()); + + // Snapshot near-P vehicles (sorted by id for stable pair iteration) + List ids = new ArrayList<>(); + List positions = new ArrayList<>(); + try (KeyValueIterator it = lastPos.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + String[] ll = kv.value.split(",", 2); + double lon = Double.parseDouble(ll[0]); + double lat = Double.parseDouble(ll[1]); + if (Haversine.withinMetres(lon, lat, pLon, pLat, dPMetres)) { + ids.add(new int[]{kv.key}); + positions.add(new double[]{lon, lat}); + } + } + } + // Sort by id for stable output (small N — bubble is fine) + int n = ids.size(); + for (int i = 0; i < n - 1; i++) { + for (int j = i + 1; j < n; j++) { + if (ids.get(i)[0] > ids.get(j)[0]) { + int[] ti = ids.get(i); ids.set(i, ids.get(j)); ids.set(j, ti); + double[] tp = positions.get(i); positions.set(i, positions.get(j)); positions.set(j, tp); + } + } + } + for (int i = 0; i < n; i++) { + for (int j = i + 1; j < n; j++) { + double d = Haversine.distanceMetres( + positions.get(i)[0], positions.get(i)[1], + positions.get(j)[0], positions.get(j)[1]); + if (d <= dMeetMetres) { + String pairKey = ids.get(i)[0] + "_" + ids.get(j)[0] + "@" + trip.getTimestamp(); + ctx.forward(new Record<>(pairKey, d, trip.getTimestamp())); + } + } + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q5SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q5SnapshotProcessor.java new file mode 100644 index 0000000..1f79ec8 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q5SnapshotProcessor.java @@ -0,0 +1,99 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q5 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, which pairs of vehicles are meeting near point P + * (using each vehicle's most-recent-known position on or before T)?" + * + *

Caller keys the input by a constant so the shared cross-vehicle + * last-known store lives in one subtask. Per event: update last-known. + * Per STREAM_TIME punctuator fire: snapshot the map, filter to near-P, + * enumerate all pairs (a { + + private final String storeName; + private final double pLon, pLat, dPMetres, dMeetMetres; + private final long snapshotTickMillis; + private KeyValueStore lastPos; // vehicleId -> "lon,lat" + private ProcessorContext ctx; + + public Q5SnapshotProcessor(String storeName, double pLon, double pLat, + double dPMetres, double dMeetMetres, + long snapshotTickMillis) { + this.storeName = storeName; + this.pLon = pLon; + this.pLat = pLat; + this.dPMetres = dPMetres; + this.dMeetMetres = dMeetMetres; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.lastPos = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + lastPos.put(trip.getVehicleId(), trip.getLon() + "," + trip.getLat()); + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List ids = new ArrayList<>(); + List positions = new ArrayList<>(); + try (KeyValueIterator it = lastPos.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + String[] ll = kv.value.split(",", 2); + double lon = Double.parseDouble(ll[0]); + double lat = Double.parseDouble(ll[1]); + if (Haversine.withinMetres(lon, lat, pLon, pLat, dPMetres)) { + ids.add(new int[]{kv.key}); + positions.add(new double[]{lon, lat}); + } + } + } + int n = ids.size(); + for (int i = 0; i < n - 1; i++) { + for (int j = i + 1; j < n; j++) { + if (ids.get(i)[0] > ids.get(j)[0]) { + int[] ti = ids.get(i); ids.set(i, ids.get(j)); ids.set(j, ti); + double[] tp = positions.get(i); positions.set(i, positions.get(j)); positions.set(j, tp); + } + } + } + for (int i = 0; i < n; i++) { + for (int j = i + 1; j < n; j++) { + double d = Haversine.distanceMetres( + positions.get(i)[0], positions.get(i)[1], + positions.get(j)[0], positions.get(j)[1]); + if (d <= dMeetMetres) { + String pairKey = ids.get(i)[0] + "_" + ids.get(j)[0] + "@" + tick; + ctx.forward(new Record<>(pairKey, d, tick)); + } + } + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q5WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q5WindowedProcessor.java new file mode 100644 index 0000000..5ae0c7f --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q5WindowedProcessor.java @@ -0,0 +1,137 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q5 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, which pairs of vehicles met near P + * (using each vehicle's last-seen-in-window position)?" + * + *

State value encodes per-window per-vehicle last position as + * {@code "vid:lon,lat|vid:lon,lat|..."}. On the STREAM_TIME punctuator + * for each closed window, filter to near-P, enumerate sorted pairs, + * forward meeting pairs. + */ +public class Q5WindowedProcessor implements Processor { + + private final String storeName; + private final double pLon, pLat, dPMetres, dMeetMetres; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q5WindowedProcessor(String storeName, double pLon, double pLat, + double dPMetres, double dMeetMetres, long windowSizeMs) { + this.storeName = storeName; + this.pLon = pLon; + this.pLat = pLat; + this.dPMetres = dPMetres; + this.dMeetMetres = dMeetMetres; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String s = winState.get(winStart); + StringBuilder rebuilt = new StringBuilder(); + boolean replaced = false; + if (s != null && !s.isEmpty()) { + for (String chunk : s.split("\\|")) { + int colon = chunk.indexOf(':'); + int vid = Integer.parseInt(chunk.substring(0, colon)); + if (vid == trip.getVehicleId()) { + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(vid).append(":").append(trip.getLon()).append(",").append(trip.getLat()); + replaced = true; + } else { + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(chunk); + } + } + } + if (!replaced) { + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(trip.getVehicleId()).append(":").append(trip.getLon()).append(",").append(trip.getLat()); + } + winState.put(winStart, rebuilt.toString()); + } + + private void punctuate(long currentStreamTime) { + List closedStarts = new ArrayList<>(); + List closedStates = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + closedStarts.add(kv.key); + closedStates.add(kv.value); + } + } + } + Integer[] idx = new Integer[closedStarts.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(closedStarts::get)); + for (Integer i : idx) { + long winStart = closedStarts.get(i); + // Parse positions, filter near-P + List nearIds = new ArrayList<>(); + List positions = new ArrayList<>(); + for (String chunk : closedStates.get(i).split("\\|")) { + int colon = chunk.indexOf(':'); + if (colon < 0) continue; + int vid = Integer.parseInt(chunk.substring(0, colon)); + String[] ll = chunk.substring(colon + 1).split(",", 2); + double lon = Double.parseDouble(ll[0]); + double lat = Double.parseDouble(ll[1]); + if (Haversine.withinMetres(lon, lat, pLon, pLat, dPMetres)) { + nearIds.add(new int[]{vid}); + positions.add(new double[]{lon, lat}); + } + } + int n = nearIds.size(); + for (int a = 0; a < n - 1; a++) { + for (int b = a + 1; b < n; b++) { + if (nearIds.get(a)[0] > nearIds.get(b)[0]) { + int[] ti = nearIds.get(a); nearIds.set(a, nearIds.get(b)); nearIds.set(b, ti); + double[] tp = positions.get(a); positions.set(a, positions.get(b)); positions.set(b, tp); + } + } + } + for (int a = 0; a < n; a++) { + for (int b = a + 1; b < n; b++) { + double d = Haversine.distanceMetres( + positions.get(a)[0], positions.get(a)[1], + positions.get(b)[0], positions.get(b)[1]); + if (d <= dMeetMetres) { + String pairKey = nearIds.get(a)[0] + "_" + nearIds.get(b)[0] + "@win" + winStart; + ctx.forward(new Record<>(pairKey, d, winStart + windowSizeMs - 1)); + } + } + } + winState.delete(winStart); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q6ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q6ContinuousProcessor.java new file mode 100644 index 0000000..6e5b9c2 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q6ContinuousProcessor.java @@ -0,0 +1,58 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; + +/** + * BerlinMOD-Q6 — continuous form, Kafka-Streams Processor API. + * + *

"What is each vehicle's cumulative distance travelled so far?" + * + *

Keyed by vehicleId. Per-vehicle state holds the last-known (lon, lat) + * and the running total in metres. On each event, accumulate the Haversine + * delta and emit the cumulative total. + * + *

State value uses a small string encoding "lon,lat,total" since the + * scaffold avoids declaring a dedicated tuple SerDe; the encoding is + * private to this processor. + * + *

TODO(meos): replace with the MEOS trajectory {@code length} call via + * the JMEOS bridge. + */ +public class Q6ContinuousProcessor implements Processor { + + private final String storeName; + private KeyValueStore state; // "lastLon,lastLat,total" + private ProcessorContext ctx; + + public Q6ContinuousProcessor(String storeName) { + this.storeName = storeName; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.state = context.getStateStore(storeName); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + String prev = state.get(trip.getVehicleId()); + double total; + if (prev == null) { + total = 0.0; + } else { + String[] parts = prev.split(",", 3); + double lastLon = Double.parseDouble(parts[0]); + double lastLat = Double.parseDouble(parts[1]); + double prevTotal = Double.parseDouble(parts[2]); + total = prevTotal + Haversine.distanceMetres(lastLon, lastLat, trip.getLon(), trip.getLat()); + } + state.put(trip.getVehicleId(), trip.getLon() + "," + trip.getLat() + "," + total); + ctx.forward(new Record<>(trip.getVehicleId(), total, trip.getTimestamp())); + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q6SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q6SnapshotProcessor.java new file mode 100644 index 0000000..4865fcd --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q6SnapshotProcessor.java @@ -0,0 +1,81 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q6 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, what is each vehicle's total distance travelled + * up to T?" + * + *

Caller keys the input by a constant. State value "lon,lat,total" + * per vehicleId. Per event: accumulate Haversine delta. Per STREAM_TIME + * punctuator fire: emit {@code (currentTick, vehicleId, total)} for + * every vehicle (sorted by vehicleId), encoded as "vid:total". + */ +public class Q6SnapshotProcessor implements Processor { + + private final String storeName; + private final long snapshotTickMillis; + private KeyValueStore state; + private ProcessorContext ctx; + + public Q6SnapshotProcessor(String storeName, long snapshotTickMillis) { + this.storeName = storeName; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.state = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + String prev = state.get(trip.getVehicleId()); + double total; + if (prev == null) { + total = 0.0; + } else { + String[] parts = prev.split(",", 3); + double lastLon = Double.parseDouble(parts[0]); + double lastLat = Double.parseDouble(parts[1]); + double prevTotal = Double.parseDouble(parts[2]); + total = prevTotal + Haversine.distanceMetres(lastLon, lastLat, trip.getLon(), trip.getLat()); + } + state.put(trip.getVehicleId(), trip.getLon() + "," + trip.getLat() + "," + total); + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List ids = new ArrayList<>(); + try (KeyValueIterator it = state.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + ids.add(kv.key); + } + } + ids.sort(Comparator.naturalOrder()); + for (Integer vid : ids) { + String[] parts = state.get(vid).split(",", 3); + double total = Double.parseDouble(parts[2]); + ctx.forward(new Record<>(tick, vid + ":" + total, tick)); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q6WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q6WindowedProcessor.java new file mode 100644 index 0000000..4d1c359 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q6WindowedProcessor.java @@ -0,0 +1,126 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q6 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, per vehicle, distance travelled + * during the window." + * + *

State encodes per-window per-vehicle {@code "vid:lastLon,lastLat,total|..."}. + * On each event, accumulate Haversine delta from the previous in-window + * position. On punctuator: emit per-vehicle totals for closed windows. + */ +public class Q6WindowedProcessor implements Processor { + + private final String storeName; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q6WindowedProcessor(String storeName, long windowSizeMs) { + this.storeName = storeName; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String s = winState.get(winStart); + StringBuilder rebuilt = new StringBuilder(); + boolean replaced = false; + if (s != null && !s.isEmpty()) { + for (String chunk : s.split("\\|")) { + int colon = chunk.indexOf(':'); + int vid = Integer.parseInt(chunk.substring(0, colon)); + String body = chunk.substring(colon + 1); + if (vid == trip.getVehicleId()) { + String[] f = body.split(",", 3); + double lastLon = Double.parseDouble(f[0]); + double lastLat = Double.parseDouble(f[1]); + double prevTotal = Double.parseDouble(f[2]); + double newTotal = prevTotal + Haversine.distanceMetres( + lastLon, lastLat, trip.getLon(), trip.getLat()); + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(vid).append(":") + .append(trip.getLon()).append(",") + .append(trip.getLat()).append(",") + .append(newTotal); + replaced = true; + } else { + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(chunk); + } + } + } + if (!replaced) { + if (rebuilt.length() > 0) rebuilt.append("|"); + rebuilt.append(trip.getVehicleId()).append(":") + .append(trip.getLon()).append(",") + .append(trip.getLat()).append(",0.0"); + } + winState.put(winStart, rebuilt.toString()); + } + + private void punctuate(long currentStreamTime) { + List closedStarts = new ArrayList<>(); + List closedStates = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + closedStarts.add(kv.key); + closedStates.add(kv.value); + } + } + } + Integer[] idx = new Integer[closedStarts.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(closedStarts::get)); + for (Integer i : idx) { + long winStart = closedStarts.get(i); + List ids = new ArrayList<>(); + List totals = new ArrayList<>(); + for (String chunk : closedStates.get(i).split("\\|")) { + int colon = chunk.indexOf(':'); + if (colon < 0) continue; + int vid = Integer.parseInt(chunk.substring(0, colon)); + String[] f = chunk.substring(colon + 1).split(",", 3); + double total = Double.parseDouble(f[2]); + ids.add(new int[]{vid}); + totals.add(new double[]{total}); + } + Integer[] sortedIdx = new Integer[ids.size()]; + for (int k = 0; k < sortedIdx.length; k++) sortedIdx[k] = k; + java.util.Arrays.sort(sortedIdx, Comparator.comparingInt(k -> ids.get(k)[0])); + for (Integer k : sortedIdx) { + ctx.forward(new Record<>(winStart, + ids.get(k)[0] + ":" + totals.get(k)[0], + winStart + windowSizeMs - 1)); + } + winState.delete(winStart); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q7ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q7ContinuousProcessor.java new file mode 100644 index 0000000..4170fc9 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q7ContinuousProcessor.java @@ -0,0 +1,60 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.util.List; + +/** + * BerlinMOD-Q7 — continuous form, Kafka-Streams Processor API. + * + *

"For each (vehicle, POI) pair, when did the vehicle first come + * within the POI's radius?" + * + *

Keyed by vehicleId. State is a {@link KeyValueStore} whose composite + * key is the POI id, scoped per-vehicle by Kafka Streams' implicit + * per-key state partitioning. On each event, scan the POI list; for each + * POI not yet recorded as passed and within radius, record the time and + * emit {@code (poiId, firstPassageTime)} (the key carries the vehicleId + * implicitly via the upstream keying). + */ +public class Q7ContinuousProcessor implements Processor { + + private final String storeName; + private final List pois; + private KeyValueStore firstPassed; // poiId -> firstPassageTime, per-vehicle by keying + private ProcessorContext ctx; + + public Q7ContinuousProcessor(String storeName, List pois) { + this.storeName = storeName; + this.pois = pois; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.firstPassed = context.getStateStore(storeName); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + Integer vehicleId = record.key(); + for (PointOfInterest poi : pois) { + int compositeKey = compositeKey(vehicleId, poi.id); + if (firstPassed.get(compositeKey) != null) continue; + if (Haversine.withinMetres(trip.getLon(), trip.getLat(), poi.lon, poi.lat, poi.radiusMetres)) { + firstPassed.put(compositeKey, trip.getTimestamp()); + ctx.forward(new Record<>(poi.id, trip.getTimestamp(), trip.getTimestamp())); + } + } + } + + /** Pack (vehicleId, poiId) into a single Integer for the store key. */ + private static int compositeKey(int vehicleId, int poiId) { + return (vehicleId * 1000) + poiId; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q7SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q7SnapshotProcessor.java new file mode 100644 index 0000000..6374a56 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q7SnapshotProcessor.java @@ -0,0 +1,91 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q7 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, for each (vehicle, POI), the first time the vehicle + * came within the POI's radius on or before T." + * + *

Caller keys the input by a constant. Store key is the composite + * {@code vehicleId * 1000 + poiId} integer with first-passage timestamp + * as value. Per event: detect any (vehicle, POI) first-passages. Per + * STREAM_TIME punctuator fire: walk store, emit each + * {@code (currentTick, "vid:poiId:firstPassageTime")} for entries + * with firstPassageTime ≤ currentTick. + */ +public class Q7SnapshotProcessor implements Processor { + + private final String storeName; + private final List pois; + private final long snapshotTickMillis; + private KeyValueStore firstPassed; // compositeKey -> firstPassageTime + private ProcessorContext ctx; + + public Q7SnapshotProcessor(String storeName, List pois, long snapshotTickMillis) { + this.storeName = storeName; + this.pois = pois; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.firstPassed = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + for (PointOfInterest poi : pois) { + int composite = trip.getVehicleId() * 1000 + poi.id; + if (firstPassed.get(composite) != null) continue; + if (Haversine.withinMetres(trip.getLon(), trip.getLat(), poi.lon, poi.lat, poi.radiusMetres)) { + firstPassed.put(composite, trip.getTimestamp()); + } + } + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List rows = new ArrayList<>(); // {vehicleId, poiId, firstPassageTime as int-of-long? — keep long via array of longs} + // separate list for longs since arrays of ints lose precision + List firstPassages = new ArrayList<>(); + try (KeyValueIterator it = firstPassed.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.value <= tick) { + int vid = kv.key / 1000; + int poiId = kv.key % 1000; + rows.add(new int[]{vid, poiId}); + firstPassages.add(kv.value); + } + } + } + // Sort by (vehicleId, poiId) for deterministic output via parallel sort of two lists + Integer[] idx = new Integer[rows.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator + .comparingInt((Integer i) -> rows.get(i)[0]) + .thenComparingInt(i -> rows.get(i)[1])); + for (Integer i : idx) { + int[] r = rows.get(i); + ctx.forward(new Record<>(tick, r[0] + ":" + r[1] + ":" + firstPassages.get(i), tick)); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q7WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q7WindowedProcessor.java new file mode 100644 index 0000000..7eb08aa --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q7WindowedProcessor.java @@ -0,0 +1,115 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q7 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, for each (vehicle, POI), the first + * event in the window where the vehicle is inside the POI." Intra- + * window scoping (no cross-window first-passage state). + * + *

State encodes per-window the recorded (vehicle, POI, time) triples + * as {@code "vid:poiId:t,vid:poiId:t,..."}. On each event scan POIs; if + * vehicle inside and this (vehicle, POI) not yet recorded for this + * window, append. Punctuator emits each recorded triple for closed + * windows. + */ +public class Q7WindowedProcessor implements Processor { + + private final String storeName; + private final List pois; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q7WindowedProcessor(String storeName, List pois, long windowSizeMs) { + this.storeName = storeName; + this.pois = pois; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String s = winState.get(winStart); + if (s == null) s = ""; + StringBuilder appended = new StringBuilder(s); + for (PointOfInterest poi : pois) { + String marker = trip.getVehicleId() + ":" + poi.id + ":"; + if (s.contains(marker)) continue; + if (Haversine.withinMetres(trip.getLon(), trip.getLat(), poi.lon, poi.lat, poi.radiusMetres)) { + if (appended.length() > 0) appended.append(","); + appended.append(trip.getVehicleId()).append(":").append(poi.id).append(":").append(trip.getTimestamp()); + } + } + if (appended.length() > s.length()) { + winState.put(winStart, appended.toString()); + } else if (s.isEmpty()) { + winState.put(winStart, ""); + } + } + + private void punctuate(long currentStreamTime) { + List closedStarts = new ArrayList<>(); + List closedStates = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + closedStarts.add(kv.key); + closedStates.add(kv.value); + } + } + } + Integer[] idx = new Integer[closedStarts.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(closedStarts::get)); + for (Integer i : idx) { + long winStart = closedStarts.get(i); + String s = closedStates.get(i); + if (s != null && !s.isEmpty()) { + List vps = new ArrayList<>(); + List times = new ArrayList<>(); + for (String chunk : s.split(",")) { + String[] f = chunk.split(":", 3); + vps.add(new int[]{Integer.parseInt(f[0]), Integer.parseInt(f[1])}); + times.add(new long[]{Long.parseLong(f[2])}); + } + Integer[] sortedIdx = new Integer[vps.size()]; + for (int k = 0; k < sortedIdx.length; k++) sortedIdx[k] = k; + java.util.Arrays.sort(sortedIdx, Comparator + .comparingInt((Integer k) -> vps.get(k)[0]) + .thenComparingInt(k -> vps.get(k)[1])); + for (Integer k : sortedIdx) { + int[] vp = vps.get(k); + ctx.forward(new Record<>(winStart, + vp[0] + ":" + vp[1] + ":" + times.get(k)[0], + winStart + windowSizeMs - 1)); + } + } + winState.delete(winStart); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q8ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q8ContinuousProcessor.java new file mode 100644 index 0000000..d910061 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q8ContinuousProcessor.java @@ -0,0 +1,45 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; + +/** + * BerlinMOD-Q8 — continuous form, Kafka-Streams Processor API. + * + *

"Is this vehicle currently within {@code d} metres of the road + * segment?" + * + *

Stateless per-event predicate using planar point-to-segment distance. + * Same shape as {@link Q3ContinuousProcessor} but with a segment-distance + * predicate instead of a point-radius one. + */ +public class Q8ContinuousProcessor implements Processor { + + private final double s1Lon, s1Lat, s2Lon, s2Lat, radiusMetres; + private ProcessorContext ctx; + + public Q8ContinuousProcessor(double s1Lon, double s1Lat, double s2Lon, double s2Lat, double radiusMetres) { + this.s1Lon = s1Lon; + this.s1Lat = s1Lat; + this.s2Lon = s2Lon; + this.s2Lat = s2Lat; + this.radiusMetres = radiusMetres; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + boolean near = SegmentDistance.withinMetres( + trip.getLon(), trip.getLat(), + s1Lon, s1Lat, s2Lon, s2Lat, + radiusMetres); + ctx.forward(new Record<>(trip.getVehicleId(), near, trip.getTimestamp())); + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q8SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q8SnapshotProcessor.java new file mode 100644 index 0000000..e6aed09 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q8SnapshotProcessor.java @@ -0,0 +1,80 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q8 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, which vehicles are within {@code d} metres of the road + * segment (using each vehicle's most-recent-known position on or before + * T)?" + * + *

Same shape as {@link Q3SnapshotProcessor} with the point-to-segment + * predicate substituted for the point-radius one. + */ +public class Q8SnapshotProcessor implements Processor { + + private final String storeName; + private final double s1Lon, s1Lat, s2Lon, s2Lat, radiusMetres; + private final long snapshotTickMillis; + private KeyValueStore lastPos; + private ProcessorContext ctx; + + public Q8SnapshotProcessor(String storeName, double s1Lon, double s1Lat, + double s2Lon, double s2Lat, double radiusMetres, + long snapshotTickMillis) { + this.storeName = storeName; + this.s1Lon = s1Lon; + this.s1Lat = s1Lat; + this.s2Lon = s2Lon; + this.s2Lat = s2Lat; + this.radiusMetres = radiusMetres; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.lastPos = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + lastPos.put(trip.getVehicleId(), trip.getLon() + "," + trip.getLat()); + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + List nearIds = new ArrayList<>(); + try (KeyValueIterator it = lastPos.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + String[] ll = kv.value.split(",", 2); + double lon = Double.parseDouble(ll[0]); + double lat = Double.parseDouble(ll[1]); + if (SegmentDistance.withinMetres(lon, lat, s1Lon, s1Lat, s2Lon, s2Lat, radiusMetres)) { + nearIds.add(kv.key); + } + } + } + nearIds.sort(Comparator.naturalOrder()); + for (Integer vid : nearIds) { + ctx.forward(new Record<>(tick, vid, tick)); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q8WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q8WindowedProcessor.java new file mode 100644 index 0000000..f36e648 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q8WindowedProcessor.java @@ -0,0 +1,103 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * BerlinMOD-Q8 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, how many distinct vehicles were + * within {@code d} metres of the road segment at any time during the + * window?" + * + *

Same shape as {@link Q3WindowedProcessor} with the segment-distance + * predicate substituted for the point-radius one. + */ +public class Q8WindowedProcessor implements Processor { + + private final String storeName; + private final double s1Lon, s1Lat, s2Lon, s2Lat, radiusMetres; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q8WindowedProcessor(String storeName, double s1Lon, double s1Lat, + double s2Lon, double s2Lat, double radiusMetres, + long windowSizeMs) { + this.storeName = storeName; + this.s1Lon = s1Lon; + this.s1Lat = s1Lat; + this.s2Lon = s2Lon; + this.s2Lat = s2Lat; + this.radiusMetres = radiusMetres; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + if (!SegmentDistance.withinMetres(trip.getLon(), trip.getLat(), + s1Lon, s1Lat, s2Lon, s2Lat, + radiusMetres)) { + return; + } + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String prior = winState.get(winStart); + String sv = Integer.toString(trip.getVehicleId()); + if (prior == null) { + winState.put(winStart, sv); + } else if (!contains(prior, sv)) { + winState.put(winStart, prior + "," + sv); + } + } + + private void punctuate(long currentStreamTime) { + List toEmit = new ArrayList<>(); + List counts = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + toEmit.add(kv.key); + Set distinct = new HashSet<>(); + for (String s : kv.value.split(",")) distinct.add(s); + counts.add((long) distinct.size()); + } + } + } + Integer[] idx = new Integer[toEmit.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(toEmit::get)); + for (Integer i : idx) { + long winStart = toEmit.get(i); + ctx.forward(new Record<>(winStart, counts.get(i), winStart + windowSizeMs - 1)); + winState.delete(winStart); + } + } + + private static boolean contains(String csv, String id) { + for (String s : csv.split(",")) if (s.equals(id)) return true; + return false; + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q9ContinuousProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q9ContinuousProcessor.java new file mode 100644 index 0000000..f26dd46 --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q9ContinuousProcessor.java @@ -0,0 +1,69 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; + +/** + * BerlinMOD-Q9 — continuous form, Kafka-Streams Processor API. + * + *

"What is the current distance between vehicles X and Y?" + * + *

Caller should pre-filter the stream to {@code vehicleId ∈ {X, Y}} and + * key by a constant so the shared X+Y state lives in a single subtask. + * State encoded as a "xLon,xLat|yLon,yLat" string with NaN sentinels for + * unseen slots; per-event update the X or Y slot then forward + * {@code (eventTime, distanceMetres)} if both slots are known. + */ +public class Q9ContinuousProcessor implements Processor { + + private static final String UNSET = "NaN,NaN"; + + private final String storeName; + private final int xVehicleId; + private final int yVehicleId; + private KeyValueStore state; // single-key (0) -> "xLon,xLat|yLon,yLat" + private ProcessorContext ctx; + + public Q9ContinuousProcessor(String storeName, int xVehicleId, int yVehicleId) { + this.storeName = storeName; + this.xVehicleId = xVehicleId; + this.yVehicleId = yVehicleId; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.state = context.getStateStore(storeName); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + String s = state.get(0); + if (s == null) { + s = UNSET + "|" + UNSET; + } + String[] parts = s.split("\\|", 2); + String xSlot = parts[0]; + String ySlot = parts[1]; + if (trip.getVehicleId() == xVehicleId) { + xSlot = trip.getLon() + "," + trip.getLat(); + } else if (trip.getVehicleId() == yVehicleId) { + ySlot = trip.getLon() + "," + trip.getLat(); + } else { + return; + } + state.put(0, xSlot + "|" + ySlot); + if (!xSlot.startsWith("NaN") && !ySlot.startsWith("NaN")) { + String[] x = xSlot.split(",", 2); + String[] y = ySlot.split(",", 2); + double d = Haversine.distanceMetres( + Double.parseDouble(x[0]), Double.parseDouble(x[1]), + Double.parseDouble(y[0]), Double.parseDouble(y[1])); + ctx.forward(new Record<>(trip.getTimestamp(), d, trip.getTimestamp())); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q9SnapshotProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q9SnapshotProcessor.java new file mode 100644 index 0000000..50c624f --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q9SnapshotProcessor.java @@ -0,0 +1,79 @@ +package berlinmod; + +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; + +/** + * BerlinMOD-Q9 — snapshot form, Kafka-Streams Processor API. + * + *

"At time T, what is the distance between vehicles X and Y (using + * their most-recent-known positions on or before T)?" + * + *

Single-key state holds the X+Y position pair. Per event for X or Y: + * update the slot. Per STREAM_TIME punctuator fire: if both slots are + * known, emit {@code (currentTick, distanceMetres)}. + */ +public class Q9SnapshotProcessor implements Processor { + + private static final String UNSET = "NaN,NaN"; + + private final String storeName; + private final int xVehicleId; + private final int yVehicleId; + private final long snapshotTickMillis; + private KeyValueStore state; // key=0 -> "xLon,xLat|yLon,yLat" + private ProcessorContext ctx; + + public Q9SnapshotProcessor(String storeName, int xVehicleId, int yVehicleId, long snapshotTickMillis) { + this.storeName = storeName; + this.xVehicleId = xVehicleId; + this.yVehicleId = yVehicleId; + this.snapshotTickMillis = snapshotTickMillis; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.state = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(snapshotTickMillis), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + String s = state.get(0); + if (s == null) s = UNSET + "|" + UNSET; + String[] parts = s.split("\\|", 2); + String xSlot = parts[0]; + String ySlot = parts[1]; + if (trip.getVehicleId() == xVehicleId) { + xSlot = trip.getLon() + "," + trip.getLat(); + } else if (trip.getVehicleId() == yVehicleId) { + ySlot = trip.getLon() + "," + trip.getLat(); + } else { + return; + } + state.put(0, xSlot + "|" + ySlot); + } + + private void punctuate(long currentStreamTime) { + long tick = (currentStreamTime / snapshotTickMillis) * snapshotTickMillis; + String s = state.get(0); + if (s == null) return; + String[] parts = s.split("\\|", 2); + if (parts[0].startsWith("NaN") || parts[1].startsWith("NaN")) return; + String[] x = parts[0].split(",", 2); + String[] y = parts[1].split(",", 2); + double d = Haversine.distanceMetres( + Double.parseDouble(x[0]), Double.parseDouble(x[1]), + Double.parseDouble(y[0]), Double.parseDouble(y[1])); + ctx.forward(new Record<>(tick, d, tick)); + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/Q9WindowedProcessor.java b/kafka-streams-app/src/main/java/berlinmod/Q9WindowedProcessor.java new file mode 100644 index 0000000..ec0ca3c --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/Q9WindowedProcessor.java @@ -0,0 +1,102 @@ +package berlinmod; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.processor.PunctuationType; +import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; +import org.apache.kafka.streams.processor.api.Record; +import org.apache.kafka.streams.state.KeyValueIterator; +import org.apache.kafka.streams.state.KeyValueStore; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** + * BerlinMOD-Q9 — windowed form, Kafka-Streams Processor API. + * + *

"Per N-second tumbling window, X-Y distance at window-end using + * each vehicle's last position in the window." + * + *

State value encodes per-window the latest X and Y positions seen + * as {@code "xLon,xLat|yLon,yLat"} (with {@code NaN,NaN} for unseen). + * Punctuator emits distance for each closed window where both X and Y + * are known. + */ +public class Q9WindowedProcessor implements Processor { + + private static final String UNSET = "NaN,NaN"; + + private final String storeName; + private final int xVehicleId, yVehicleId; + private final long windowSizeMs; + private KeyValueStore winState; + private ProcessorContext ctx; + + public Q9WindowedProcessor(String storeName, int xVehicleId, int yVehicleId, long windowSizeMs) { + this.storeName = storeName; + this.xVehicleId = xVehicleId; + this.yVehicleId = yVehicleId; + this.windowSizeMs = windowSizeMs; + } + + @Override + public void init(ProcessorContext context) { + this.ctx = context; + this.winState = context.getStateStore(storeName); + context.schedule(Duration.ofMillis(windowSizeMs), + PunctuationType.STREAM_TIME, this::punctuate); + } + + @Override + public void process(Record record) { + BerlinMODTrip trip = record.value(); + if (trip == null || trip.getVehicleId() == -1) return; + long winStart = (trip.getTimestamp() / windowSizeMs) * windowSizeMs; + String s = winState.get(winStart); + if (s == null) s = UNSET + "|" + UNSET; + String[] parts = s.split("\\|", 2); + String xSlot = parts[0]; + String ySlot = parts[1]; + if (trip.getVehicleId() == xVehicleId) { + xSlot = trip.getLon() + "," + trip.getLat(); + } else if (trip.getVehicleId() == yVehicleId) { + ySlot = trip.getLon() + "," + trip.getLat(); + } else { + return; + } + winState.put(winStart, xSlot + "|" + ySlot); + } + + private void punctuate(long currentStreamTime) { + List closedStarts = new ArrayList<>(); + List closedStates = new ArrayList<>(); + try (KeyValueIterator it = winState.all()) { + while (it.hasNext()) { + KeyValue kv = it.next(); + if (kv.key + windowSizeMs <= currentStreamTime) { + closedStarts.add(kv.key); + closedStates.add(kv.value); + } + } + } + Integer[] idx = new Integer[closedStarts.size()]; + for (int i = 0; i < idx.length; i++) idx[i] = i; + java.util.Arrays.sort(idx, Comparator.comparingLong(closedStarts::get)); + for (Integer i : idx) { + long winStart = closedStarts.get(i); + String s = closedStates.get(i); + String[] parts = s.split("\\|", 2); + if (!parts[0].startsWith("NaN") && !parts[1].startsWith("NaN")) { + String[] x = parts[0].split(",", 2); + String[] y = parts[1].split(",", 2); + double d = Haversine.distanceMetres( + Double.parseDouble(x[0]), Double.parseDouble(x[1]), + Double.parseDouble(y[0]), Double.parseDouble(y[1])); + ctx.forward(new Record<>(winStart, d, winStart + windowSizeMs - 1)); + } + winState.delete(winStart); + } + } +} diff --git a/kafka-streams-app/src/main/java/berlinmod/SegmentDistance.java b/kafka-streams-app/src/main/java/berlinmod/SegmentDistance.java new file mode 100644 index 0000000..295d42e --- /dev/null +++ b/kafka-streams-app/src/main/java/berlinmod/SegmentDistance.java @@ -0,0 +1,54 @@ +package berlinmod; + +/** + * Distance from a (lon, lat) point to a (lon, lat) line segment, in metres, + * via a local equirectangular projection centred on the segment midpoint. + * + *

Same shape as the MobilityFlink {@code SegmentDistance} utility so + * BerlinMOD-Q8 produces byte-identical cross-platform output. TODO(meos): + * replace with MEOS {@code distance(tgeompoint, geometry(LINESTRING))} + * via the JMEOS bridge. + */ +public final class SegmentDistance { + + private static final double EARTH_RADIUS_METRES = 6_371_000.0; + + private SegmentDistance() {} + + public static double distanceMetres( + double pLon, double pLat, + double s1Lon, double s1Lat, + double s2Lon, double s2Lat) { + double midLat = (s1Lat + s2Lat) / 2.0; + double mPerDegLat = Math.toRadians(1.0) * EARTH_RADIUS_METRES; + double mPerDegLon = mPerDegLat * Math.cos(Math.toRadians(midLat)); + + double px = pLon * mPerDegLon; + double py = pLat * mPerDegLat; + double s1x = s1Lon * mPerDegLon; + double s1y = s1Lat * mPerDegLat; + double s2x = s2Lon * mPerDegLon; + double s2y = s2Lat * mPerDegLat; + + double dx = s2x - s1x; + double dy = s2y - s1y; + double lenSq = dx * dx + dy * dy; + if (lenSq == 0.0) { + return Math.hypot(px - s1x, py - s1y); + } + double t = ((px - s1x) * dx + (py - s1y) * dy) / lenSq; + if (t < 0.0) t = 0.0; + else if (t > 1.0) t = 1.0; + double cx = s1x + t * dx; + double cy = s1y + t * dy; + return Math.hypot(px - cx, py - cy); + } + + public static boolean withinMetres( + double pLon, double pLat, + double s1Lon, double s1Lat, + double s2Lon, double s2Lat, + double radiusMetres) { + return distanceMetres(pLon, pLat, s1Lon, s1Lat, s2Lon, s2Lat) <= radiusMetres; + } +} diff --git a/kafka-streams-app/src/main/resources/log4j.properties b/kafka-streams-app/src/main/resources/log4j.properties new file mode 100644 index 0000000..641e984 --- /dev/null +++ b/kafka-streams-app/src/main/resources/log4j.properties @@ -0,0 +1,13 @@ +# Minimal log4j config for the MobilityKafka scaffold + +log4j.rootLogger=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} [%t] %-5p %c{1} - %m%n + +log4j.logger.berlinmod=INFO + +# Kafka-Streams internals at WARN to keep test output readable +log4j.logger.org.apache.kafka=WARN +log4j.logger.org.apache.kafka.streams=WARN diff --git a/kafka-streams-app/src/test/java/berlinmod/BerlinMODFullMatrixTest.java b/kafka-streams-app/src/test/java/berlinmod/BerlinMODFullMatrixTest.java new file mode 100644 index 0000000..dfa8114 --- /dev/null +++ b/kafka-streams-app/src/test/java/berlinmod/BerlinMODFullMatrixTest.java @@ -0,0 +1,183 @@ +package berlinmod; + +import org.apache.kafka.common.serialization.BooleanDeserializer; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.DoubleDeserializer; +import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.TestInputTopic; +import org.apache.kafka.streams.TopologyTestDriver; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * End-to-end regression test for the full BerlinMOD-9 × 3-form Kafka-Streams + * parity matrix (all 27 cells, not just Q1). + * + *

Runs the whole {@link BerlinMODTopology} in-process via + * {@link TopologyTestDriver}, pipes a slice of the canonical BerlinMOD corpus plus two + * sentinel events that advance stream-time past the snapshot/windowed + * punctuator boundaries, and asserts the per-{@code Q}-form output-record + * count for every continuous, windowed and snapshot topic. + */ +class BerlinMODFullMatrixTest { + + private static final Logger LOG = LoggerFactory.getLogger(BerlinMODFullMatrixTest.class); + + @Test + void fullMatrix() { + LOG.info("BerlinMODFullMatrixTest starting (TopologyTestDriver, all 27 cells)"); + + Properties props = new Properties(); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, "berlinmod-test"); + 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(); + + try (TopologyTestDriver driver = new TopologyTestDriver(BerlinMODTopology.build(), props)) { + + TestInputTopic input = + driver.createInputTopic(BerlinMODTopology.INPUT_TOPIC, + new IntegerSerializer(), + tripSerde.serializer()); + + List corpus = loadCanonicalSample(); + for (BerlinMODTrip trip : corpus) { + input.pipeInput(trip.getVehicleId(), trip, + Instant.ofEpochMilli(trip.getTimestamp())); + } + // Sentinel events (vehicleId == -1, ignored by all processors) to advance + // stream-time past the final snapshot/windowed punctuator boundaries after + // the last canonical instant. Two steps because the STREAM_TIME punctuator + // coalesces a multi-interval jump into a single fire. + long lastTs = corpus.get(corpus.size() - 1).getTimestamp(); + for (long s : new long[]{lastTs + 6_000L, lastTs + 12_000L}) { + input.pipeInput(-1, new BerlinMODTrip(-1, s, 0.0, 0.0), Instant.ofEpochMilli(s)); + } + + // Q5 ("pairs of vehicles meeting near P") is the one cell whose count + // differs between the pure-Java Haversine approximation (scaffold) and + // the MEOS geodetic engine (integration): the MEOS pair-meeting path + // emits none under these params while the planar approximation is more + // permissive. The other 26 cells are engine-invariant. TODO(meos): + // reconcile the Q5 pair-meeting semantics across the two engines. + boolean meos = Boolean.getBoolean("meos.enabled"); + + // ---- continuous outputs ---- + assertCount(driver, BerlinMODTopology.Q1_CONTINUOUS_OUTPUT, "Q1-continuous", 5, + Serdes.Integer().deserializer(), new LongDeserializer()); + assertCount(driver, BerlinMODTopology.Q2_CONTINUOUS_OUTPUT, "Q2-continuous", 40, + Serdes.Integer().deserializer(), tripSerde.deserializer()); + assertCount(driver, BerlinMODTopology.Q3_CONTINUOUS_OUTPUT, "Q3-continuous", 200, + Serdes.Integer().deserializer(), new BooleanDeserializer()); + assertCount(driver, BerlinMODTopology.Q4_CONTINUOUS_OUTPUT, "Q4-continuous", 2, + Serdes.Integer().deserializer(), new LongDeserializer()); + assertCount(driver, BerlinMODTopology.Q5_CONTINUOUS_OUTPUT, "Q5-continuous", meos ? 0 : 199, + new StringDeserializer(), new DoubleDeserializer()); + assertCount(driver, BerlinMODTopology.Q6_CONTINUOUS_OUTPUT, "Q6-continuous", 200, + Serdes.Integer().deserializer(), new DoubleDeserializer()); + assertCount(driver, BerlinMODTopology.Q7_CONTINUOUS_OUTPUT, "Q7-continuous", 3, + Serdes.Integer().deserializer(), new LongDeserializer()); + assertCount(driver, BerlinMODTopology.Q8_CONTINUOUS_OUTPUT, "Q8-continuous", 200, + Serdes.Integer().deserializer(), new BooleanDeserializer()); + assertCount(driver, BerlinMODTopology.Q9_CONTINUOUS_OUTPUT, "Q9-continuous", 79, + new LongDeserializer(), new DoubleDeserializer()); + + // ---- windowed outputs ---- + assertCount(driver, BerlinMODTopology.Q1_WINDOWED_OUTPUT, "Q1-windowed", 14, + new LongDeserializer(), new LongDeserializer()); + assertCount(driver, BerlinMODTopology.Q2_WINDOWED_OUTPUT, "Q2-windowed", 7, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q3_WINDOWED_OUTPUT, "Q3-windowed", 12, + new LongDeserializer(), new LongDeserializer()); + assertCount(driver, BerlinMODTopology.Q4_WINDOWED_OUTPUT, "Q4-windowed", 21, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q5_WINDOWED_OUTPUT, "Q5-windowed", meos ? 0 : 7, + new StringDeserializer(), new DoubleDeserializer()); + assertCount(driver, BerlinMODTopology.Q6_WINDOWED_OUTPUT, "Q6-windowed", 46, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q7_WINDOWED_OUTPUT, "Q7-windowed", 24, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q8_WINDOWED_OUTPUT, "Q8-windowed", 11, + new LongDeserializer(), new LongDeserializer()); + assertCount(driver, BerlinMODTopology.Q9_WINDOWED_OUTPUT, "Q9-windowed", 7, + new LongDeserializer(), new DoubleDeserializer()); + + // ---- snapshot outputs ---- + assertCount(driver, BerlinMODTopology.Q1_SNAPSHOT_OUTPUT, "Q1-snapshot", 141, + new LongDeserializer(), new IntegerDeserializer()); + assertCount(driver, BerlinMODTopology.Q2_SNAPSHOT_OUTPUT, "Q2-snapshot", 28, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q3_SNAPSHOT_OUTPUT, "Q3-snapshot", 29, + new LongDeserializer(), new IntegerDeserializer()); + assertCount(driver, BerlinMODTopology.Q4_SNAPSHOT_OUTPUT, "Q4-snapshot", 57, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q5_SNAPSHOT_OUTPUT, "Q5-snapshot", meos ? 0 : 28, + new StringDeserializer(), new DoubleDeserializer()); + assertCount(driver, BerlinMODTopology.Q6_SNAPSHOT_OUTPUT, "Q6-snapshot", 141, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q7_SNAPSHOT_OUTPUT, "Q7-snapshot", 84, + new LongDeserializer(), new StringDeserializer()); + assertCount(driver, BerlinMODTopology.Q8_SNAPSHOT_OUTPUT, "Q8-snapshot", 56, + new LongDeserializer(), new IntegerDeserializer()); + assertCount(driver, BerlinMODTopology.Q9_SNAPSHOT_OUTPUT, "Q9-snapshot", 28, + new LongDeserializer(), new DoubleDeserializer()); + } + LOG.info("BerlinMODFullMatrixTest done — all 27 cells matched expected counts"); + } + + private static void assertCount( + TopologyTestDriver driver, String topic, String tag, int expected, + Deserializer kd, Deserializer vd) { + int actual = driver.createOutputTopic(topic, kd, vd).readKeyValuesToList().size(); + LOG.info("{} output: {} lines (expected {})", tag, actual, expected); + LOG.info("{} output: {} lines (expected {})", tag, actual, expected); + assertEquals(expected, actual, tag + " output-record count"); + } + + /** + * Loads the canonical BerlinMOD sample (real instants for vehicles 1-5, + * reprojected to WGS84 and time-aligned to a common base) from the test + * resources. This is a slice of the single canonical BerlinMOD corpus + * shared across the ecosystem — no invented coordinates. + */ + private static List loadCanonicalSample() { + List events = new ArrayList<>(); + try (BufferedReader r = new BufferedReader(new InputStreamReader( + BerlinMODFullMatrixTest.class.getResourceAsStream("/berlinmod_sample.csv"), + StandardCharsets.UTF_8))) { + r.readLine(); // header: vehicleId,timestampMs,lon,lat + for (String line; (line = r.readLine()) != null; ) { + String[] f = line.split(","); + events.add(new BerlinMODTrip(Integer.parseInt(f[0].trim()), + Long.parseLong(f[1].trim()), + Double.parseDouble(f[2].trim()), + Double.parseDouble(f[3].trim()))); + } + } catch (Exception e) { + throw new RuntimeException("loading /berlinmod_sample.csv", e); + } + // Sort by event-time so all vehicles' early instants are seen before the + // first punctuator tick, matching a real Kafka source's monotonic delivery. + events.sort(Comparator.comparingLong(BerlinMODTrip::getTimestamp)); + return events; + } +} diff --git a/kafka-streams-app/src/test/resources/berlinmod_sample.csv b/kafka-streams-app/src/test/resources/berlinmod_sample.csv new file mode 100644 index 0000000..43477c3 --- /dev/null +++ b/kafka-streams-app/src/test/resources/berlinmod_sample.csv @@ -0,0 +1,201 @@ +vehicleId,timestampMs,lon,lat +1,1735711200000,4.4348828,50.7678369 +2,1735711200000,4.3321436,50.7689294 +3,1735711200000,4.4565674,50.8519902 +4,1735711200000,4.3044542,50.8819037 +5,1735711200000,4.4251697,50.9188357 +1,1735711201500,4.4348387,50.7678314 +2,1735711201500,4.3321424,50.7689578 +3,1735711201500,4.4565913,50.8520142 +4,1735711201500,4.3044987,50.8819077 +5,1735711201500,4.4251937,50.9188596 +2,1735711202250,4.3321412,50.7689862 +5,1735711202250,4.4252178,50.9188835 +3,1735711202350,4.4566139,50.8520369 +1,1735711202400,4.4347947,50.7678258 +5,1735711202750,4.4252418,50.9189075 +5,1735711203125,4.4252658,50.9189314 +5,1735711203425,4.4252899,50.9189553 +4,1735711203642,4.3045431,50.8819118 +3,1735711204150,4.4567037,50.8520381 +3,1735711204441,4.4567182,50.8520383 +4,1735711205651,4.3045489,50.8819123 +2,1735711206450,4.332133,50.769185 +2,1735711206458,4.332133,50.769185 +2,1735711206575,4.3321329,50.7691872 +4,1735711207018,4.3045914,50.8819214 +5,1735711207024,4.4256264,50.9192902 +2,1735711207325,4.332128,50.7692154 +2,1735711207925,4.332123,50.7692437 +5,1735711208615,4.4256264,50.9192902 +4,1735711208818,4.3046765,50.8819395 +3,1735711208941,4.4567011,50.8518969 +4,1735711209116,4.3046773,50.8819397 +5,1735711209401,4.425639,50.9193027 +3,1735711209425,4.4566993,50.8518817 +5,1735711210151,4.4256629,50.9193267 +1,1735711210331,4.4347681,50.7678225 +4,1735711210493,4.3047168,50.8819532 +5,1735711210651,4.4256868,50.9193506 +5,1735711211026,4.4257107,50.9193746 +5,1735711211326,4.4257346,50.9193986 +2,1735711211393,4.3321204,50.7692585 +1,1735711211677,4.4347269,50.7678111 +2,1735711212616,4.3321031,50.7692847 +3,1735711213025,4.4566871,50.8517685 +2,1735711213289,4.3320858,50.7693109 +2,1735711214356,4.3320858,50.7693109 +2,1735711215694,4.3320703,50.7693343 +5,1735711215697,4.4261409,50.9198062 +3,1735711216025,4.456684,50.8517402 +5,1735711216340,4.4261648,50.9198301 +2,1735711216444,4.3320558,50.7693612 +5,1735711216790,4.4261887,50.9198541 +4,1735711216793,4.3049936,50.8820473 +2,1735711216936,4.3320439,50.7693832 +3,1735711217025,4.456681,50.851712 +5,1735711217136,4.4262126,50.9198781 +5,1735711217417,4.4262365,50.9199021 +2,1735711217536,4.3320294,50.7694101 +3,1735711217925,4.4566779,50.8516837 +5,1735711218189,4.4263082,50.919974 +3,1735711218404,4.4566763,50.8516686 +5,1735711218430,4.4263306,50.9199965 +4,1735711219191,4.3049936,50.8820473 +4,1735711219891,4.305012,50.8820536 +3,1735711220204,4.4566735,50.8516119 +5,1735711220230,4.4262131,50.920038 +1,1735711221577,4.434274,50.7676863 +1,1735711221762,4.434274,50.7676863 +1,1735711221824,4.4342723,50.7676858 +5,1735711221945,4.4261739,50.9200519 +3,1735711222004,4.456672,50.8515836 +5,1735711222442,4.4261495,50.9200605 +3,1735711222904,4.4566706,50.8515552 +3,1735711222921,4.4566706,50.8515547 +5,1735711222964,4.4261278,50.9200357 +5,1735711223351,4.4261061,50.9200109 +5,1735711223659,4.4260844,50.9199861 +3,1735711223821,4.4566989,50.8515327 +5,1735711224151,4.4260428,50.9199387 +1,1735711225424,4.4341128,50.7676335 +3,1735711225799,4.4567154,50.8515199 +2,1735711225844,4.3320162,50.7694345 +2,1735711227133,4.3320107,50.7694627 +1,1735711227674,4.4340729,50.7676204 +2,1735711230354,4.3320052,50.7694909 +1,1735711230374,4.4339533,50.7675812 +1,1735711230561,4.433945,50.7675785 +4,1735711230691,4.305519,50.8821691 +4,1735711231305,4.305519,50.8821691 +4,1735711232542,4.3055538,50.882177 +4,1735711233442,4.305596,50.8821866 +4,1735711234166,4.30563,50.8821944 +3,1735711234301,4.4567602,50.8515183 +5,1735711234694,4.4250675,50.918954 +5,1735711234907,4.4250478,50.9189341 +5,1735711235164,4.425024,50.9189101 +5,1735711235296,4.4250118,50.9188977 +3,1735711235576,4.4568051,50.8515167 +2,1735711241103,4.3320015,50.7695097 +3,1735711241876,4.457119,50.8515054 +2,1735711242476,4.3320012,50.7695381 +1,1735711242859,4.4333967,50.7674041 +4,1735711243166,4.3060608,50.8822746 +2,1735711243192,4.332001,50.7695665 +3,1735711244503,4.4571294,50.851505 +5,1735711245325,4.4240805,50.9179624 +2,1735711245592,4.3319999,50.7696801 +3,1735711245827,4.4571722,50.8514964 +5,1735711246059,4.4240567,50.9179384 +2,1735711246107,4.3319999,50.7696801 +5,1735711246553,4.4240328,50.9179144 +5,1735711246924,4.4240089,50.9178904 +2,1735711247187,4.3319997,50.7697006 +5,1735711247221,4.423985,50.9178664 +2,1735711247937,4.3320014,50.769729 +3,1735711249427,4.4573435,50.8514622 +2,1735711252137,4.332013,50.7699277 +2,1735711252477,4.332013,50.7699277 +3,1735711252549,4.4573435,50.8514622 +2,1735711253656,4.3320143,50.76995 +3,1735711254049,4.4573863,50.8514536 +2,1735711254406,4.3320164,50.7699784 +4,1735711255151,4.3060987,50.8822817 +1,1735711255459,4.4328196,50.7672461 +4,1735711256502,4.3061382,50.8822952 +1,1735711257259,4.4327784,50.7672348 +2,1735711258006,4.3320289,50.7701486 +5,1735711258278,4.4229583,50.9168351 +5,1735711258430,4.4229442,50.916821 +3,1735711259449,4.4576432,50.8514022 +3,1735711260101,4.4576742,50.851396 +2,1735711260277,4.3320295,50.7701561 +2,1735711261554,4.3320413,50.7701835 +3,1735711261901,4.4577607,50.8513807 +2,1735711262244,4.332053,50.7702109 +3,1735711262627,4.4577956,50.8513745 +3,1735711263190,4.4577865,50.8513467 +3,1735711263599,4.4577773,50.851319 +3,1735711263959,4.4577682,50.8512912 +3,1735711264245,4.4577609,50.8512691 +1,1735711265359,4.4324075,50.7671332 +2,1735711265423,4.3320572,50.7702206 +1,1735711265453,4.4324075,50.7671332 +4,1735711265502,4.3065335,50.8824297 +3,1735711265685,4.4577408,50.8511564 +1,1735711266001,4.4323924,50.7671291 +2,1735711266709,4.3320777,50.7702459 +2,1735711267401,4.3320982,50.7702711 +5,1735711267687,4.4220808,50.9159591 +4,1735711267692,4.306572,50.8824428 +2,1735711268011,4.3320982,50.7702711 +2,1735711269214,4.3321147,50.7702914 +2,1735711269964,4.3321363,50.7703163 +5,1735711271825,4.4220808,50.9159591 +3,1735711274590,4.4577366,50.8511331 +4,1735711274892,4.3068147,50.88261 +4,1735711275650,4.3068402,50.8826276 +3,1735711275907,4.4577363,50.8511047 +3,1735711276608,4.4577359,50.8510764 +1,1735711280401,4.4317184,50.7669714 +4,1735711281050,4.3070414,50.8827407 +1,1735711282201,4.4316762,50.7669616 +3,1735711282202,4.4577358,50.8510646 +3,1735711283552,4.4577392,50.8510363 +4,1735711283621,4.307075,50.8827595 +4,1735711284569,4.3071085,50.8827784 +1,1735711284901,4.4315499,50.766932 +1,1735711287472,4.4315077,50.7669222 +1,1735711288420,4.4314656,50.7669123 +4,1735711292669,4.3074104,50.882948 +4,1735711292919,4.3074104,50.882948 +4,1735711293060,4.3074135,50.8829498 +4,1735711293960,4.3074423,50.8829716 +4,1735711295245,4.3074711,50.8829933 +4,1735711296145,4.3074999,50.8830151 +4,1735711296991,4.3075269,50.8830355 +4,1735711297891,4.3075557,50.8830573 +4,1735711298608,4.3075786,50.8830746 +4,1735711299198,4.3076042,50.8830838 +4,1735711301198,4.3076433,50.8830978 +4,1735711302322,4.3076433,50.8830978 +4,1735711302734,4.307654,50.8831016 +1,1735711302820,4.4307916,50.7667547 +1,1735711303090,4.4307789,50.7667517 +1,1735711306690,4.4306122,50.7667093 +4,1735711307234,4.3078317,50.8831882 +1,1735711308551,4.4306047,50.7667074 +1,1735711309806,4.4305706,50.7666908 +1,1735711310754,4.4305531,50.766663 +1,1735711311654,4.4305504,50.7666346 +1,1735711313282,4.4305498,50.7666283 +1,1735711314525,4.4305643,50.7666014 +1,1735711318125,4.4306223,50.7664939 +1,1735711319144,4.4306223,50.7664939 +1,1735711320162,4.4306321,50.7664756 +1,1735711327362,4.4307639,50.7662642 +1,1735711328862,4.4307804,50.7662377 +1,1735711340562,4.4309945,50.7658942 +1,1735711342525,4.4309984,50.7658879 +1,1735711343795,4.4310279,50.7658665