Spark Streaming Kafka 10 API Word Count application Scala

In Spark Kafka Streaming Java program Word Count using Kafka 0.10 API blog entry i talked about how you create a simple java program that uses Spark Streaming's Kafka10 API using Java. This blog entry does the same thing but using Scala. You can download the complete application from github
You can run this sample by first downloading Kafka 0.10.* from Apache Kafka WebSite, then you can create and start a test topic and send messages to it by following this Kafka Quick start document

Spark Kafka Streaming Java program Word Count using Kafka 0.10 API

Kafka API went through a lot of changes starting Kafka 0.9. Spark Kafka Streaming API also was changed to better support Kafka 0.9. i wanted to try that out so i built this simple Word Count application using Kafka 0.10 API. This blog entry does the same thing but using Scala. You can download the complete application from github
You can run this sample by first downloading Kafka 0.10.* from Apache Kafka WebSite, then you can create and start a test topic and send messages to it by following this Kafka Quick start document First thing i did was to include Kafka 0.10 API dependencies for the Spark Project. As you can see i am using Spark 2.1 version Then i did create a SparkKafka10.java file that looks like this. Please take a look at comments inside the code for what i am doing. Now if you create test topic and send messages to it, you should be able to see the wordcount on console

Flume to Spark Streaming - Pull model

In this post i will demonstrate how to stream data from flume into Spark using Streaming. When it comes to Streaming data from Flume to Spark you have 2 options.

1. Push Model: Spark listens on particular port for Avro event and flume connects to that port and publishes event
2. Pull Model: You use special Spark Sink in flume that keeps collecting published data and Spark pulls that data at certain frequency

I built this simple configuration in which i could send event to flume on netcat, flume would take those events and send them to Spark as well as print to console.

• First download spark-streaming-flume-sink_2.10-1.6.0.jar and copy it to flume/lib directory
• Next create flume configuration that looks like this, as you can see, Flume is listening for netcat event on port 44444 and it is taking every event and replicating it to both logger and Spark sink. Spark sink would listen on port 9999 for Spark program to connect
• This is how your Spark driver will look like. The Spark Flume listener gets event in avro format so you will have to call event.getBody().array() to get the event.

Once your spark and flume agents are started open netcat on port 44444 and send messages, those messages should appear in your Spark Console

Monitoring HDFS directory for new files using Spark Streaming

I wanted to build this simple Spark Streaming application that monitors a particular directory in HDFS and whenever a new file shows up, i want to print its content to Console. I built this HDFSFileStream.scala. In this program after creating a SparkStreamContext. I am calling sparkStreamingContext.textFileStream(<directoryName>) on it. Once a new file appears in the directory the value of fileRDD.count() would return more than 0 and then i invoke processNewFile(). The processNewFile() method takes a RDD[String], iterates through the file content and prints it to console Next start the program by executing following code bin/spark-submit ~/HelloSparkStreaming-1.0-SNAPSHOT-jar-with-dependencies.jar /user/mapr/stream 3 Once the streaming started it starts monitoring /user/mapr/stream directory, for new content. I copied a file with few lines in it and i got the following output, which is content of the file

Hello Spark Streaming

In the WordCount program built using Apache Spark in Java , i built simple Spark program that takes name of the file as input, reads the file and performs word count on the file. Now Spark also has concept of Spark Streaming which allows you to read file as stream of real time events instead of one time load of input file. But the API for transforming the data, in both cases Spark converts the input in RDD. In case of Spark Streaming it would convert the input events into Micro RDD which is nothing but collecting all the incoming data for certain duration (microBatchTime) and then exposes it as RDD. I built this simple NetcatStreamClient Streaming application that listens for incoming data on netcat port, once it has data it performs wordcount on it and prints that to console. You can download the full source code from GitHub

package com.spnotes.spark

import com.typesafe.scalalogging.Logger
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.slf4j.LoggerFactory

object NetcatStreamClient{
  val logger = Logger(LoggerFactory.getLogger("NetcatStreamClient"))

  def main(argv:Array[String]): Unit ={
    logger.debug("Entering NetcatStreamClient.main")
    if(argv.length != 3){
      println("Please provide 3 parameters   ")
      System.exit(1)
    }
    val hostName =argv(0)
    val port = argv(1).toInt
    val microBatchTime = argv(2).toInt

    logger.debug(s"Listening on $hostName at $port batching records every $microBatchTime")

    //Create Spark Configuration
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
 
 
    //Create SparkStreamingContext with microBatchTime which specifies how long spark should collect data
    val sparkStreamingContext = new StreamingContext(sparkConf,Seconds(microBatchTime))
    //Start listening for data on given host and port
    val lines = sparkStreamingContext.socketTextStream(hostName,port)
// Logic for implementing word count on the input batch
    lines.flatMap(_.split(" ")).map(word => (word,1)).reduceByKey(_+_).print()
 
    logger.debug("Number of words " + lines.count())

    //Start the stream so that data starts flowing, you must define transformation logic before calling start()
    sparkStreamingContext.start()
    sparkStreamingContext.awaitTermination()
    logger.debug("Exiting NetcatStreamClient.main")
  }
}

Once your project is built using mvn clean compile assembly:single, first thing you should do is executing following command to start netcat on localhost at port 9999 nc -l 9999

Next execute following code to start Spark Driver that takes 3 parameters host and port where netcat is listening and last parameter is how the batch duration should be bin/spark-submit ~/HelloSparkStreaming-1.0-SNAPSHOT-jar-with-dependencies.jar localhost 9999 5