Suppose product prices arrive continuously and you need to select expensive items, convert their prices, or route them to different topics. A stream processor applies one operation to records as they flow through a pipeline. Kafka Streams and Apache Flink connect processors so that input can become filtered, transformed, or aggregated output. section 12.2
Choose a processor for the job:
- Read: `builder.stream("products.prices.changelog")` creates a stream from a Kafka topic. section 12.2.1
- Select: `filter()` keeps records whose predicate is true, such as prices above $100. section 12.2.1
- Transform: `map()` can change both key and value; `mapValues()` changes only the value. section 12.2.1
- Combine or route: `merge()` combines streams; `split()` sends records down branches selected by conditions. section 12.2.1
- Count: `groupByKey().count()` counts records by key. Counting is stateful because it retains intermediate results. section 12.2.1 section 12.7
mapValues((key, value) -> value.price * 100)
This `mapValues()` example converts a price from dollars to cents without changing the record key. Prefer it over `map()` when the key should stay the same: changing keys can require repartitioning and reduce performance. section 12.2.1
prices.split() #1
.branch((key, value) -> value.price < 10, #2
Branched.withConsumer( #3
(stream) -> stream.to("prices-cheap"))) #4
.branch((key, value) -> value.price < 100, #5
Branched.withConsumer(
(stream) -> stream.to("prices-midrange")))
.defaultBranch(Branched.withConsumer( #6
(stream) -> stream.to("prices-expensive")))
The price stream branches into cheap, midrange, and expensive outputs, each written to its own Kafka topic. It shows how connected processors turn one incoming stream into several useful outputs. section 12.2.1
Connect processors into a topology for longer workflows, such as merging streams before filtering or aggregating them. Unlike a simple filter or conversion, an aggregation must maintain state. Stream processing also requires accounting for eventual consistency; key-value stores and changelog topics help protect stored state against data loss. section 12.7
