Scala Kafka Consumer

In this post we will learn how to get messages from Kafka using Scala.

Add the following library dependency.

libraryDependencies += "org.apache.kafka" %% "kafka" % "2.1.0"

Following is the scala code to get (consume) messages from Kafka.

import java.util






import org.apache.kafka.clients.consumer.KafkaConsumer






import java.util.Properties






import scala.collection.JavaConverters._













class GetMessagesFromKafka {













  def main(args: Array[String]): Unit = {






    getMessagesFromKafka("Topic1")






  }













  def getMessagesFromKafka(topic: String) = {






    val props = new Properties()






    props.put("bootstrap.servers", "localhost:9094")






    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")






    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")






    props.put("auto.offset.reset", "latest")






    props.put("group.id", "consumer-group")






    val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)






    consumer.subscribe(util.Arrays.asList(topic))






    while (true) {






      val record = consumer.poll(1000).asScala






      for (data <- record.iterator)






        println(data.value())






    }






  }






}

That's it!

Cheers!

Scala Kafka Producer

In this post we will learn how to send messages to Kafka using Scala.

Add the following library dependency.

libraryDependencies += "org.apache.kafka" %% "kafka" % "2.1.0"

Following is the scala code to send (produce) messages to Kafka.

import java.util.Properties






import org.apache.kafka.clients.producer._













object SendMessagesToKafka {













  def main(args: Array[String]): Unit = {






    sendMessagesToKafka("Topic1")






  }













  def sendMessagesToKafka(topic: String): Unit = {






    val props = new Properties()






    props.put("bootstrap.servers", "localhost:9094")






    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")






    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")






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






    val record = new ProducerRecord[String, String](topic, "key", "value")






    producer.send(record)






    producer.close()






  }






}

That's it!

Cheers!

Convert AVRO to ORC using Scala

In this article we will see how to convert an AVRO file to a ORC file using a Spark Dataframe using Scala.

The input AVRO file has been taken from: https://github.com/Teradata/kylo/blob/master/samples/sample-data/avro/userdata1.avro

The SBT library dependencies are shown below for reference.

scalaVersion := "2.11.12"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.3.0"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.3.0"

libraryDependencies += "com.databricks" %% "spark-avro" % "3.2.0"

The Scala program is provided below.

import org.apache.spark.sql.{SaveMode, SparkSession}

object AvroToORCConverter extends App {

  System.setProperty("hadoop.home.dir","C:\\intellij.winutils")

  val spark = SparkSession.builder()
    .master("local")
    .appName("AvroToORCConverter")
    .config("spark.sql.orc.impl", "native")
    .getOrCreate()

  val inputFile = "C:\\data\\data.avro"  val outputFile = "C:\\data\\out_data_avro2orc"
  import spark.implicits._

  val df = spark    .read
    .format("com.databricks.spark.avro")
    .load(inputFile)

  df    .write
    .mode(SaveMode.Overwrite)
    .option("header","true")
    .orc(outputFile)
}

The converted ORC file is shown below.



That's all!

Convert AVRO to Parquet using Scala

In this article we will see how to convert an AVRO file to a Parquet file using a Spark Dataframe using Scala.

The input AVRO file has been taken from: https://github.com/Teradata/kylo/blob/master/samples/sample-data/avro/userdata1.avro


The SBT library dependencies are shown below for reference.

scalaVersion := "2.11.12"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.3.0"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.3.0"

libraryDependencies += "com.databricks" %% "spark-avro" % "3.2.0"

The Scala program is provided below.

import org.apache.spark.sql.{SaveMode, SparkSession}

object AvroToParquetConverter extends App {

  System.setProperty("hadoop.home.dir","C:\\intellij.winutils")

  val spark = SparkSession.builder()
    .master("local")
    .appName("AvroToParquetConverter")
    .getOrCreate()

  val inputFile = "C:\\data\\data.avro"  val outputFile = "C:\\data\\out_data_avro2parquet"
  import spark.implicits._

  val df = spark    .read
    .format("com.databricks.spark.avro")
    .load(inputFile)

  df    .write
    .mode(SaveMode.Overwrite)
    .option("header","true")
    .parquet(outputFile)

}

The converted Parquet file is shown below.



That's all!