Stream Building

Question 1

Create a SERDE for a class WindTurbine

Answer 1

ObjectMapperSerde turbineSerde = new ObjectMapperSerde<>(WindTurbine.class);

Question 2

read the “turbines” topic as a KTable

Answer 2

builder.table(
“turbines”,

Consumed.with(Serdes.Integer(), turbineSerde),

Materialized.[Integer, WindTurbine, KeyValueStore[Bytes, byte[]]as(“turbine-store”)

.withKeySerde(Serdes.Integer())

.withValueSerde(turbineSerde)
);

Question 3

read the “turbine-generated-watts” topic as a KStream

Answer 3

KStream wattsValuesStream = builder.stream(
“turbine-generated-watts”,
Consumed.with(Serdes.Integer(), Serdes.Integer())
);

Question 4

map the watts stream into a new mwatts stream

Answer 4

wattsValuesStream.map((id, watts) -> {

return KeyValue.pair(id, MWatts(turbineId, watts / 1000000));
}).to(

“turbine-generated-mwatts”,
Produced.with(Serdes.Integer(), mwattsSerde)
);

Question 5

count measurements by turbine in stream wattsValuesStream and write results to a new stream

Answer 5

wattsValuesStream
.groupByKey()
.count()
.toStream()
.map((turbineId, count) -> {
WindTurbineStats stats = new WindTurbineStats(turbineId, count);
return KeyValue.pair(turbineId, stats);
})
.to(
“turbine-stats”,
Produced.with(Serdes.Integer(), statsSerde)
);

Question 6

Kafka properties for running a stream

Answer 6

quarkus. kafka-streams.application-id
quarkus. kafka-streams.bootstrap-servers
quarkus. kafka-streams.topics

Question 7

Filter stream and write all events with amount greater than 100

Answer 7

stream
.filter((key, amount) -> amount > 1000)
.to(“large-payments”, Produced.with(keySerde, valueSerde));

Question 8

Convert a stream of distances in meters to cm

Answer 8

stream
.map((key, value) -> {
Double cm = value * 0.1;
return KeyValue.pair(key, cm);
})
.to(“vehicle-feet-elevations”, Produced.with(Serdes.Integer(), Serdes.Double()));

Question 9

Stream has vehicle positions, Keys are the color of the vehicle. group these vehicles by their Id

Answer 9

KGroupedStream positionsByVehicle = stream
.groupBy(
(key, value) -> value.vehicleId,
Grouped.with(Serdes.Integer(),vehiclePositionSerde)
);

Question 10

what is the result of the count() operation and how to you convert it to a stream

Answer 10

KTable, call toStream()

Question 11

Group users by country in a stream where the key is the language

Answer 11

userStreams.groupBy((key, value) -> value.country).

Question 12

Country users per country in a stream where the key is the language

Answer 12

userStreams.groupBy((key, value) -> value.country).count();

Question 13

Read from topic keyed by language and write to topic keyed by locationId to achieve repartitioning

Answer 13

stream.map(
(key, measure) ->new KeyValue<>(measure.locationId, measure);

).to(REPARTITIONED_TOPIC,
Produced.with(serde1, serd2)
);

Question 14

Skeleton of a Stream processor in plain Java

Answer 14