KStream.to writes records from a KStream to a Kafka topic through a sink node. It completes a stream-processing path rather than providing another KStream for chaining, so it is a terminal operation alongside print and foreach. section 6.4.2 section 6.5
A typical sink operation combines these pieces:
- Transform purchase records into a KStream, such as purchase-pattern records keyed by ZIP code. section 6.4.2
- Call to with the destination topic name, such as patterns. section 6.4.2
- Pass Produced.with using the key and value Serde required to serialize output records. section 6.4.2
- Use print or peek separately when you need development-time visibility before or after the sink path. section 6.5
Writing purchase patterns
A retailer can flatten each transaction into purchased items, map those items into Pattern values, and write the resulting stream to the patterns topic: patternKStream.to("patterns", Produced.with(stringSerde, purchasePatternSerde));. Printing the stream before this call provides console feedback, but to ends the chain. section 6.4.2 section 6.5
For per-record routing, pass a TopicNameExtractor instead of a fixed topic. Its extract method can inspect the key, value, or RecordContext, including timestamps, offsets, partitions, headers, and the source topic. Extracted destination topics must already exist; Kafka Streams does not create them by default. section 6.6.4 section 6.7
Listing 6.11 Patterns processor and sink node that writes to Kafka
KStream<String, Pattern> patternKStream = retailPurchaseKStream
.flatMap(retailTransactionToPurchases) #1
.mapValues(patternObjectMapper); #2
patternKStream.print(Printed.<String,Pattern>toSysOut()
.withLabel("patterns-stream")); #3
patternKStream.to("patterns",
Produced.with(stringSerde,purchasePatternSerde)); #4
Follow the complete pattern-stream construction from flattening and value mapping through console inspection and the Kafka sink. section 6.4.2
