learning kafka streams with scala

Learning Kafka Streams with Scala

Himanshu GuptaLead ConsultantKnoldus Software LLP

Img src - http://danlebrero.com/2017/01/05/proof-of-concept-using-kafkastreams-and-ktables/

Agenda

● What is Kafka Streams?● Why do we need to implement it in Scala?● Challenge● Solution● Examples of Kafka Streams with Scala● Demo

What is Kafka Streams?

The easiest way to write mission-critical real-time applications and microservices with all the benefits of Kafka's server-side cluster

technology

● Write standard Java applications.

● No separate processing cluster required

● Elastic, highly scalable, fault-tolerant

● Equally viable for small, medium, & large use cases

● Exactly-once processing semantics

● Develop on Mac, Linux, Windows

● Deploy to containers, VM(s), bare metal, cloud

● Fully integrated with Kafka security

Key Features

Where we can use it?

Img src - https://kafka.apache.org/documentation/streams/

Example

object WordCountApplication {     def main(args: Array[String]) {        val textLines: KStream[String, String]  =  builder.stream("TextLinesTopic")                val wordCounts: KTable[String, Long] = textLines            .flatMapValues(textLine => textLine.toLowerCase.split("\\W+").toIterable.asJava)            .groupBy((_, word) => word)            .count("Counts")        wordCounts.to(Serdes.String(), Serdes.Long(), "WordsWithCountsTopic")         val streams: KafkaStreams = new KafkaStreams(builder, config)        streams.start()    } }

Why do we need to implement it in Scala?

● To provide a functional solution to a problem.

● To get better concurrency model.

● To write succinct and concise code.

● To have type safety or static typing.

Challenge

No Scala API

Challenge

No Scala API

Is it really a challenge?

Challenge

No Scala API

Is it really a challenge?

Yes

Challenge

No Scala API

Is it really a challenge?

Yes

Because Scala does not accept a function literal as a valid expression for any Single Abstract Method (SAM) type

Example

Scala functions don’t implement the Runnable and Callable interfaces even though there’s a simply mapping.

val r: Runnable = () => println("Run!")

error: type mismatch;

 found   : () => Unit

 required: Runnable

  val r: Runnable = () => println("Run!")

Compile

Solution???

Solution

val r: Runnable =    new Runnable() {       def run() = println("Run!")    }

Explicitly implement Runnable interface anonymously in Scala

Solution

val r: Runnable =    new Runnable() {       def run() = println("Run!")    }

Explicitly implement Runnable interface anonymously in Scala

What’s the problem with this approach?

Solution

val r: Runnable =    new Runnable() {       def run() = println("Run!")    }

Explicitly implement Runnable interface anonymously in Scala

Its too verbose!

What can be other solution?

Alternative Solution

Use Scala 2.12

Alternative Solution

Use Scala 2.12

But why?

Alternative Solution

Use Scala 2.12

But why?

Because Scala 2.12 accepts a function literal as a valid expression for any Single Abstract Method (SAM) type

Example

It works!

It doesn’t work!

Key Features of Scala 2.12

● A trait compiles directly to an interface with default methods. This improves binary compatibility and Java interoperability.

● Scala and Java 8 interop is also improved for functional code, as methods that take functions can easily be called in both directions using lambda syntax. The FunctionN classes in Scala’s standard library are now Single Abstract Method (SAM) types, and all SAM types are treated uniformly – from type checking through code generation. No class file is generated for a lambda, invokedynamic is used instead.

Examples of Kafka Streams with Scala

Example #1 – Simple Pipeline

Example #2 – Mapped Pipeline (w/o SAM)

Example #2 – Mapped Pipeline (with SAM)

Example #2 – Aggregation Pipeline (w/o SAM)

Example #2 – Aggregation Pipeline (with SAM)

There are other Examples too - https://github.com/knoldus/kafka-streams-scala-examples

DEMO

References

● https://kafka.apache.org/documentation/streams/● http://www.scala-lang.org/news/2.12.0/#lambda-syntax-for-sam-types

Thank You