The first big step to work with Kafka is to put data in a topic, and so is the purpose of this post. Being JSON the most common way to intercomunicate, and having the schema with the data, we will explore how we build a producer in scala to start populating our system.

NOTE
I am using MapR so not all configurations are the same in its Open Source counterparts (e.g. I don't need to add the ZooKeeper nodes)

Generate random input data

The first step to simulate a data pipeline will be to generate a more complex data, for example in JSON. For this, we'll use the json4s library that allows to easily convert a case class to a JSON string. Other options would be to use jackson or play JSON. However, since I want to explore other JSON libraries, this has been the choosen one.

case class PassengersDiff(event: String,
                          trainId: Int,
                          line: String,
                          stopTime: Timestamp,
                          passengerOn: Int,
                          passengersOff: Int)

val trainLines = List("R1", "R2", "R3", "R4", "RT1", "L3", "S2", "L7", "RL2", "L5", "L1", "S1")
def randomLine: String = trainLines(Random.nextInt(trainLines.size))
def randomPassengerCount: Int = Random.nextInt(100)
def randomTrainNumber: Int = Random.nextInt(6)

def randomDiff: PassengersDiff = PassengersDiff(
    UUID.randomUUID().toString,
    randomTrainNumber,
    randomLine,
    new Timestamp(System.currentTimeMillis()),
    randomPassengerCount,
    randomPassengerCount
)

Once we have the code to generate a case class with random values, it can be converted to a JSON string with the following piece of code.

import org.json4s.native.Serialization
import org.json4s.native.Serialization.write


implicit val formats = Serialization.formats(NoTypeHints)
val value = write(randomDiff)

Kafka producer: put put put

Once we have our data in a string format, we need to initializate a Producer so we can send the data. The configurations are similar to the ones from kafka stremas, but instend of having a default serializer, we'll only use one for the key and another for the value. In both our cases we'll use strings.

val TOPIC_OUTPUT = "/steampipe:json-data"

val kafkaCfg = {
  val settings = new Properties()
  settings.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  settings.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  settings
}

val producer = new KafkaProducer[String, String](kafkaCfg)

And once we have the producer, we will continuouslly put data to the topic. Easy peasy.

while (true) {
  implicit val formats = Serialization.formats(NoTypeHints)
  val value = write(randomDiff)
  // Send the key and data
  producer.send(new ProducerRecord(TOPIC_OUTPUT, passengerDiff.line, value))
  Thread.sleep(Random.nextInt(1000))
}

And we can see what this will look like in the topic

[mapr@maprdemo ~]$ /opt/mapr/kafka/kafka-1.1.1/bin/kafka-console-consumer.sh --bootstrap-server fake.9092 --topic /steampipe:json-data --property print.key=true
R3	{"event":"9bf57c73-3169-4114-867d-41740e28e949","trainId":5,"line":"R3","stopTime":"2019-08-23T18:57:34.118Z","passengerOn":80,"passengersOff":73}
RL2	{"event":"4723c00a-6549-41cf-9722-634526188d47","trainId":5,"line":"RL2","stopTime":"2019-08-23T18:57:35.286Z","passengerOn":14,"passengersOff":34}
L1	{"event":"93797a8a-aa00-452c-89fb-5a9c9659b7c7","trainId":4,"line":"L1","stopTime":"2019-08-23T18:57:35.847Z","passengerOn":24,"passengersOff":62}
L7	{"event":"a9dd38a1-aecb-4a8f-a472-63c751d62ceb","trainId":1,"line":"L7","stopTime":"2019-08-23T18:57:36.310Z","passengerOn":81,"passengersOff":49}

Now, other applications can consume from it, and we are all set to move to explore avro, finding a couple nuences along the way.

The kafka series

You can check out the previous post in the series

Or the next post
--> Using Avro in Kafka streams: the hard way

As always, you can find the source code in my github repository (see the code for JSONProducer).