Skip to content
This repository was archived by the owner on May 25, 2023. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 71 additions & 9 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,15 +1,77 @@
# Kafka Streams Goodies for Scala developers
# A Thin Scala Wrapper Around the Kafka Streams Java API

This repository contains the following:
The library wraps Java APIs in Scala thereby providing:

1. [Scala APIs for Kafka Streams](https://github.com/lightbend/kafka-streams-scala/blob/develop/kafka-stream-s/README.md): This is a thin wrapper on top of Java APIs to provide less boilerplates and better type inference.
2. [An http layer for Kafka Streams Interactive Queries](https://github.com/lightbend/kafka-streams-scala/blob/develop/kafka-stream-q/README.md): This is a utility that's quite useful for developing global queries across local states in a Kafka Streams application. More useful when the application is deployed in a distributed manner across multiple nodes.
3. [An example application](https://github.com/lightbend/kafka-streams-scala/blob/develop/kafka-stream-q-example-dsl/README.md) based on Kafka Streams DSL that uses the library in (2).
1. much better type inference in Scala
2. less boilerplate in application code
3. the usual builder-style composition that developers get with the original Java API

These tools support Kafka 1.0.0. By default, they build for Scala 2.12 (with Scala 2.12.4), but you can build targets for both 2.12 and 2.11 (using Scala 2.11.11) in SBT by adding a plus, `+`, before each command. For example:
The design of the library was inspired by the work started by Alexis Seigneurin in [this repository](https://github.com/aseigneurin/kafka-streams-scala).

## Quick Start

`kafka-streams-scala` is published and cross-built for Scala `2.11`, and `2.12`, so you can just add the following to your build:

```scala
val kafka_streams_scala_version = "0.0.1"

libraryDependencies ++= Seq("com.lightbend" %%
"kafka-streams-scala" % kafka_streams_scala_version)
```
$ sbt
> +clean
> +publishLocal

> Note: `kafka-streams-scala` supports Kafka Streams `1.0.0`.

## Running the Tests

The library comes with an embedded Kafka server. To run the tests, simply run `sbt testOnly` and all tests will run on the local embedded server.

> The embedded server is started and stopped for every test and takes quite a bit of resources. Hence it's recommended that you allocate more heap space to `sbt` when running the tests. e.g. `sbt -mem 1500`.

## Type Inference and Composition

Here's a sample code fragment using the Scala wrapper library. Compare this with the Scala code from the same [example](https://github.com/confluentinc/kafka-streams-examples/blob/4.0.0-post/src/test/scala/io/confluent/examples/streams/StreamToTableJoinScalaIntegrationTest.scala) in Confluent's repository.

```scala
// Compute the total per region by summing the individual click counts per region.
val clicksPerRegion: KTableS[String, Long] = userClicksStream

// Join the stream against the table.
.leftJoin(userRegionsTable, (clicks: Long, region: String) => (if (region == null) "UNKNOWN" else region, clicks))

// Change the stream from <user> -> <region, clicks> to <region> -> <clicks>
.map((_, regionWithClicks) => regionWithClicks)

// Compute the total per region by summing the individual click counts per region.
.groupByKey(Serialized.`with`(stringSerde, longSerde))
.reduce(_ + _)
```

> **Note:** The left quotes around "with" are there because `with` is a Scala keyword. This is the mechanism you use to "escape" a Scala keyword when it's used as a normal identifier in a Java library.

## Better Abstraction

The wrapped Scala APIs also incur less boilerplate by taking advantage of Scala function literals that get converted to Java objects in the implementation of the API. Hence the surface syntax of the client API looks simpler and less noisy.

Here's an example of a snippet built using the Java API from Scala ..

```scala
val approximateWordCounts: KStream[String, Long] = textLines
.flatMapValues(value => value.toLowerCase.split("\\W+").toIterable.asJava)
.transform(
new TransformerSupplier[Array[Byte], String, KeyValue[String, Long]] {
override def get() = new ProbabilisticCounter
},
cmsStoreName)
approximateWordCounts.to(outputTopic, Produced.`with`(Serdes.String(), longSerde))
```

And here's the corresponding snippet using the Scala library. Note how the noise of `TransformerSupplier` has been abstracted out by the function literal syntax of Scala.

```scala
textLines
.flatMapValues(value => value.toLowerCase.split("\\W+").toIterable)
.transform(() => new ProbabilisticCounter, cmsStoreName)
.to(outputTopic, Produced.`with`(Serdes.String(), longSerde))
```

Also, the explicit conversion `asJava` from a Scala `Iterable` to a Java `Iterable` is done for you by the Scala library.
File renamed without changes.
191 changes: 0 additions & 191 deletions kafka-stream-q-example-dsl/README.md

This file was deleted.

Loading