in this chapter, we will be discussing about how to integrate apache kafka with spark streaming api.

about spark

spark streaming api enables scalable, high-throughput, fault-tolerant stream processing of live data streams. data can be ingested from many sources like kafka, flume, twitter, etc., and can be processed using complex algorithms such as high-level functions like map, reduce, join and window. finally, processed data can be pushed out to filesystems, databases, and live dash-boards. resilient distributed datasets (rdd) is a fundamental data structure of spark. it is an immutable distributed collection of objects. each dataset in rdd is divided into logical partitions, which may be computed on different nodes of the cluster.

integration with spark

kafka is a potential messaging and integration platform for spark streaming. kafka act as the central hub for real-time streams of data and are processed using complex algorithms in spark streaming. once the data is processed, spark streaming could be publishing results into yet another kafka topic or store in hdfs, databases or dashboards. the following diagram depicts the conceptual flow.

now, let us go through kafka-spark apis in detail.

sparkconf api

it represents configuration for a spark application. used to set various spark parameters as key-value pairs.

sparkconf class has the following methods −

set(string key, string value) − set configuration variable.
remove(string key) − remove key from the configuration.
setappname(string name) − set application name for your application.
get(string key) − get key

streamingcontext api

this is the main entry point for spark functionality. a sparkcontext represents the connection to a spark cluster, and can be used to create rdds, accumulators and broadcast variables on the cluster. the signature is defined as shown below.

public streamingcontext(string master, string appname, duration batchduration, 
   string sparkhome, scala.collection.seq<string> jars, 
   scala.collection.map<string,string> environment)

master − cluster url to connect to (e.g. mesos://host:port, spark://host:port, local[4]).
appname − a name for your job, to display on the cluster web ui
batchduration − the time interval at which streaming data will be divided into batches

public streamingcontext(sparkconf conf, duration batchduration)

create a streamingcontext by providing the configuration necessary for a new sparkcontext.

conf − spark parameters
batchduration − the time interval at which streaming data will be divided into batches

kafkautils api

kafkautils api is used to connect the kafka cluster to spark streaming. this api has the signifi-cant method createstream signature defined as below.

public static receiverinputdstream<scala.tuple2<string,string>> createstream(
   streamingcontext ssc, string zkquorum, string groupid,
   scala.collection.immutable.map<string,object> topics, storagelevel storagelevel)

the above shown method is used to create an input stream that pulls messages from kafka brokers.

ssc − streamingcontext object.
zkquorum − zookeeper quorum.
groupid − the group id for this consumer.
topics − return a map of topics to consume.
storagelevel − storage level to use for storing the received objects.

kafkautils api has another method createdirectstream, which is used to create an input stream that directly pulls messages from kafka brokers without using any receiver. this stream can guarantee that each message from kafka is included in transformations exactly once.

the sample application is done in scala. to compile the application, please download and install sbt, scala build tool (similar to maven). the main application code is presented below.

import java.util.hashmap

import org.apache.kafka.clients.producer.{kafkaproducer, producerconfig, produc-errecord}
import org.apache.spark.sparkconf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._

object kafkawordcount {
   def main(args: array[string]) {
      if (args.length < 4) {
         system.err.println("usage: kafkawordcount <zkquorum><group> <topics> <numthreads>")
         system.exit(1)
      }

      val array(zkquorum, group, topics, numthreads) = args
      val sparkconf = new sparkconf().setappname("kafkawordcount")
      val ssc = new streamingcontext(sparkconf, seconds(2))
      ssc.checkpoint("checkpoint")

      val topicmap = topics.split(",").map((_, numthreads.toint)).tomap
      val lines = kafkautils.createstream(ssc, zkquorum, group, topicmap).map(_._2)
      val words = lines.flatmap(_.split(" "))
      val wordcounts = words.map(x => (x, 1l))
         .reducebykeyandwindow(_ &plus; _, _ - _, minutes(10), seconds(2), 2)
      wordcounts.print()

      ssc.start()
      ssc.awaittermination()
   }
}

build script

the spark-kafka integration depends on the spark, spark streaming and spark kafka integration jar. create a new file build.sbt and specify the application details and its dependency. the sbt will download the necessary jar while compiling and packing the application.

name := "spark kafka project"
version := "1.0"
scalaversion := "2.10.5"

librarydependencies += "org.apache.spark" %% "spark-core" % "1.6.0"
librarydependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0"
librarydependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0"

compilation / packaging

run the following command to compile and package the jar file of the application. we need to submit the jar file into the spark console to run the application.

sbt package

submiting to spark

start kafka producer cli (explained in the previous chapter), create a new topic called my-first-topic and provide some sample messages as shown below.

another spark test message

run the following command to submit the application to spark console.

/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming
-kafka_2.10:1.6.0 --class "kafkawordcount" --master local[4] target/scala-2.10/spark
-kafka-project_2.10-1.0.jar localhost:2181 <group name> <topic name> <number of threads>

the sample output of this application is shown below.

spark console messages ..
(test,1)
(spark,1)
(another,1)
(message,1)
spark console message ..

Save or share this article