Spark Streaming 接入 CKafka

这里使用 0.10.2.1 版本的 Kafka 依赖。

1. 在 build.sbt 添加依赖：
   

   name := "Producer Example"
   version := "1.0"
   scalaVersion := "2.11.8"
   libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.10.2.1"

2. 配置 producer_example.scala：
   
   import java.util.Properties
   import org.apache.kafka.clients.producer._
   object ProducerExample extends App {
    val  props = new Properties()
    props.put("bootstrap.servers", "172.16.16.12:9092") //实例信息中的内网 IP 与端口
     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 TOPIC="test"  //指定要生产的 Topic
    for(i<- 1 to 50){
           val record = new ProducerRecord(TOPIC, "key", s"hello $i") //生产 key 是"key",value 是 hello i 的消息
           producer.send(record)
    }
    val record = new ProducerRecord(TOPIC, "key", "the end "+new java.util.Date)
    producer.send(record)
    producer.close() //最后要断开
   }

更多有关 ProducerRecord 的用法请参见 ProducerRecord 文档。

DirectStream

1. 在 build.sbt 添加依赖：
   

   name := "Consumer Example"
   version := "1.0"
   scalaVersion := "2.11.8"
   libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
   libraryDependencies += "org.apache.spark" %% "spark-streaming" % "2.1.0"
   libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.1.0"

2. 配置 DirectStream_example.scala：
   

   import org.apache.kafka.clients.consumer.ConsumerRecord
   import org.apache.kafka.common.serialization.StringDeserializer
   import org.apache.kafka.common.TopicPartition
   import org.apache.spark.streaming.kafka010._
   import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
   import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
   import org.apache.spark.streaming.kafka010.KafkaUtils
   import org.apache.spark.streaming.kafka010.OffsetRange
   import org.apache.spark.streaming.{Seconds, StreamingContext}
   import org.apache.spark.SparkConf
   import org.apache.spark.SparkContext
   import collection.JavaConversions._
   import Array._
   object Kafka {
       def main(args: Array[String]) {
           val kafkaParams = Map[String, Object](
               "bootstrap.servers" -> "172.16.16.12:9092",
               "key.deserializer" -> classOf[StringDeserializer],
               "value.deserializer" -> classOf[StringDeserializer],
               "group.id" -> "spark_stream_test1",
               "auto.offset.reset" -> "earliest",
               "enable.auto.commit" -> "false"
           )
   
           val sparkConf = new SparkConf()
           sparkConf.setMaster("local")
           sparkConf.setAppName("Kafka")
           val ssc = new StreamingContext(sparkConf, Seconds(5))
           val topics = Array("spark_test")
   
           val offsets : Map[TopicPartition, Long] = Map()
   
           for (i <- 0 until 3){
               val tp = new TopicPartition("spark_test", i)
               offsets.updated(tp , 0L)
           }
           val stream = KafkaUtils.createDirectStream[String, String](
               ssc,
               PreferConsistent,
               Subscribe[String, String](topics, kafkaParams)
           )
           println("directStream")
           stream.foreachRDD{ rdd=>
               //输出获得的消息
               rdd.foreach{iter =>
                   val i = iter.value
                   println(s"${i}")
               }
               //获得offset
               val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
               rdd.foreachPartition { iter =>
                   val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
                   println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
               }
           }
   
           // Start the computation
           ssc.start()
           ssc.awaitTermination()
       }
   }

RDD

1. 配置build.sbt（配置同上，单击查看）。
2. 配置RDD_example：
   
   import org.apache.kafka.clients.consumer.ConsumerRecord
   import org.apache.kafka.common.serialization.StringDeserializer
   import org.apache.spark.streaming.kafka010._
   import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
   import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
   import org.apache.spark.streaming.kafka010.KafkaUtils
   import org.apache.spark.streaming.kafka010.OffsetRange
   import org.apache.spark.streaming.{Seconds, StreamingContext}
   import org.apache.spark.SparkConf
   import org.apache.spark.SparkContext
   import collection.JavaConversions._
   import Array._
   object Kafka {
       def main(args: Array[String]) {
           val kafkaParams = Map[String, Object](
               "bootstrap.servers" -> "172.16.16.12:9092",
               "key.deserializer" -> classOf[StringDeserializer],
               "value.deserializer" -> classOf[StringDeserializer],
               "group.id" -> "spark_stream",
               "auto.offset.reset" -> "earliest",
               "enable.auto.commit" -> (false: java.lang.Boolean)
           )
           val sc = new SparkContext("local", "Kafka", new SparkConf())
           val java_kafkaParams : java.util.Map[String, Object] = kafkaParams
           //按顺序向 parition 拉取相应 offset 范围的消息，如果拉取不到则阻塞直到超过等待时间或者新生产消息达到拉取的数量
           val offsetRanges = Array[OffsetRange](
               OffsetRange("spark_test", 0, 0, 5),
               OffsetRange("spark_test", 1, 0, 5),
               OffsetRange("spark_test", 2, 0, 5)
           )
           val range = KafkaUtils.createRDD[String, String](
               sc,
               java_kafkaParams,
               offsetRanges,
               PreferConsistent
           )
           range.foreach(rdd=>println(rdd.value))
           sc.stop()
       }
   }

更多 kafkaParams 用法参见 kafkaParams 文档。