Apache Kafka

https://kafka.apache.org/documentation
Admin
Apache Kafka is an event streaming platform that enables publishing, storing, and processing streams...
Tokens:3,358
Snippets:9
Trust Score:10
Update:1 year ago
Show doc for...
Tokens:
Raw
### Kafka Setup and Topic Creation Shell Commands

Source: https://kafka.apache.org/documentation/streams/quickstart

This section provides shell commands for setting up Apache Kafka and preparing the necessary topics for the Kafka Streams demo. It includes downloading Kafka, starting the server, creating input and output topics with specific configurations, and describing the created topics.

```Shell
$ tar -xzf kafka_2.13-4.1.0.tgz
$ cd kafka_2.13-4.1.0
```

```Shell
$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
```

```Shell
$ bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties
```

```Shell
$ bin/kafka-server-start.sh config/server.properties
```

```Shell
$ bin/kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --replication-factor 1 \
    --partitions 1 \
    --topic streams-plaintext-input
```

```Shell
$ bin/kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --replication-factor 1 \
    --partitions 1 \
    --topic streams-wordcount-output \
    --config cleanup.policy=compact
```

```Shell
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe
```

--------------------------------

### Start Kafka Console Producer

Source: https://kafka.apache.org/documentation/streams/quickstart

Starts a console producer to send input data to the 'streams-plaintext-input' topic for the WordCount demo. Requires a Kafka bootstrap server address.

```bash
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic streams-plaintext-input
```

--------------------------------

### Start Kafka WordCount Demo

Source: https://kafka.apache.org/documentation/streams/quickstart

Starts the Kafka Streams WordCount demo application. This application reads from 'streams-plaintext-input', processes messages using the WordCount algorithm, and writes results to 'streams-wordcount-output'. Output is written back to Kafka, not to standard output.

```bash
bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo
```

--------------------------------

### Kafka Streams WordCount Java Example

Source: https://kafka.apache.org/documentation/streams/quickstart

This Java code snippet demonstrates the core logic of a Kafka Streams application for word counting. It defines serializers/deserializers, constructs a KStream from an input topic, performs word splitting and counting, and outputs the results to another topic.

```Java
final Serde<String> stringSerde = Serdes.String();
final Serde<Long> longSerde = Serdes.Long();

KStream<String, String> textLines = builder.stream(
      "streams-plaintext-input",
      Consumed.with(stringSerde, stringSerde)
    );

KTable<String, Long> wordCounts = textLines
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, value) -> value)
    .count();

wordCounts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));
```

--------------------------------

### Consume WordCount Output

Source: https://kafka.apache.org/documentation/streams/quickstart

Starts a console consumer to read output from the 'streams-wordcount-output' topic. It reads from the beginning of the topic and specifies key/value deserializers for String and Long types, respectively.

```bash
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic streams-wordcount-output \
    --from-beginning \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
```

--------------------------------

### Produce messages to Kafka topic

Source: https://kafka.apache.org/documentation/streams/quickstart

Uses the Kafka console producer to send lines of text to the 'streams-plaintext-input' topic. Each line is treated as a message value with a null key.

```bash
$ bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic streams-plaintext-input
>all streams lead to kafka
>hello kafka streams
>join kafka summit
```

--------------------------------

### Consume processed data from Kafka topic

Source: https://kafka.apache.org/documentation/streams/quickstart

Uses the Kafka console consumer to read messages from the 'streams-wordcount-output' topic. It's configured to print keys and values, deserializing them as String and Long respectively, to show word counts.

```bash
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic streams-wordcount-output \
    --from-beginning \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
```

--------------------------------

### Implement WordCount with Kafka Streams in Scala

Source: https://kafka.apache.org/documentation/streams

This Scala code provides a WordCount application using Kafka Streams. It mirrors the Java functionality, processing text from Kafka, counting words, and outputting results. It leverages Scala's functional programming features and requires Kafka Streams libraries.

```Scala
import java.util.Properties
import java.util.concurrent.TimeUnit

import org.apache.kafka.streams.kstream.Materialized
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}

object WordCountApplication extends App {
  import Serdes._

  val props: Properties = {
    val p = new Properties()
    p.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application")
    p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092")
    p
  }

  val builder: StreamsBuilder = new StreamsBuilder
  val textLines: KStream[String, String] = builder.stream[String, String]("TextLinesTopic")
  val wordCounts: KTable[String, Long] = textLines
    .flatMapValues(textLine => textLine.toLowerCase.split("\\W+"))
    .groupBy((_, word) => word)
    .count()(Materialized.as("counts-store"))
  wordCounts.toStream.to("WordsWithCountsTopic")

  val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
  streams.start()

  sys.ShutdownHookThread {
     streams.close(10, TimeUnit.SECONDS)
  }
}
```

--------------------------------

### Implement WordCount with Kafka Streams in Java

Source: https://kafka.apache.org/documentation/streams

This Java code implements a WordCount application using Kafka Streams. It processes text lines from a Kafka topic, counts word occurrences, and outputs the counts to another topic. It requires Kafka client libraries and Kafka Streams API.

```Java
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.Arrays;
import java.util.Properties;

public class WordCountApplication {

   public static void main(final String[] args) throws Exception {
       Properties props = new Properties();
       props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
       props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092");
       props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
       props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

       StreamsBuilder builder = new StreamsBuilder();
       KStream<String, String> textLines = builder.stream("TextLinesTopic");
       KTable<String, Long> wordCounts = textLines
           .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
           .groupBy((key, word) -> word)
           .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
       wordCounts.toStream().to("WordsWithCountsTopic", Produced.with(Serdes.String(), Serdes.Long()));

       KafkaStreams streams = new KafkaStreams(builder.build(), props);
       streams.start();
   }

}
```
Tokens:
Raw