sparkStreaming 读kafka的数据

目标：sparkStreaming每2s中读取一次kafka中的数据，进行单词计数。

topic:topic1

broker list:192.168.1.126:9092,192.168.1.127:9092,192.168.1.128:9092

1、首先往一个topic中实时生产数据。

　　代码如下： 代码功能：每秒向topic1发送一条消息，一条消息里包含4个单词，单词之间用空格隔开。

package kafkaProducer

import java.util.HashMap

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


object KafkaProducer {
def main(args: Array[String]) {
  val topic="topic1"
  val brokers="192.168.1.126:9092,192.168.1.127:9092,192.168.1.128:9092"
  val messagesPerSec=1 //每秒发送几条信息  
  val wordsPerMessage =4 //一条信息包括多少个单词  
  // Zookeeper connection properties  
    val props = new HashMap[String, Object]()  
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)  
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,  
      "org.apache.kafka.common.serialization.StringSerializer")  
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,  
      "org.apache.kafka.common.serialization.StringSerializer")  
    val producer = new KafkaProducer[String, String](props) 
    // Send some messages  
     while(true) {  
      (1 to messagesPerSec.toInt).foreach { messageNum =>  
        val str = (1 to wordsPerMessage.toInt).map(x => scala.util.Random.nextInt(10).toString)  
          .mkString(" ")  
        val message = new ProducerRecord[String, String](topic, null, str)  
        producer.send(message)  
        println(message)  
      }  
      Thread.sleep(1000)  
    }  
  }  
}

 打包运行命令：hadoop jar jar包  （注意jar包是可运行的jar包）

消费者消费命令： ./kafka-console-consumer.sh  --zookeeper zk01:2181,zk02:2181  --topic topic1 --from-beginning

可以正常消费。

2、编写SparkStreaming代码读kafka中的数据，每2s读一次

　　代码如下：

package kafkaSparkStream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils
import kafka.serializer.StringDecoder
/**
 * sparkStreaming读取kafka中topic的数据
 */
object KafkaToSpark {
def main(args: Array[String]) {
  if (args.length<2) {
  System.err.println("Usage: <brokers> <topics>");
  System.exit(1)
  }
  val Array(brokers,topics)=args
  //2s从kafka中读取一次
  val conf=new SparkConf().setAppName("KafkaToSpark");
  val scc=new StreamingContext(conf,Seconds(2))
  // Create direct kafka stream with brokers and topics  
  val topicSet=topics.split(",").toSet
  val kafkaParams=Map[String,String]("metadata.broker.list"->brokers)
  //获取信息
  val messages=KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      scc,kafkaParams,topicSet)
  // Get the lines, split them into words, count the words and print 
 val lines= messages.map(_._2)
 val words=lines.flatMap(_.split(" "))
 val wordCouts=words.map(x =>(x,1L)).reduceByKey(_+_)
 wordCouts.print
 //开启计算
 scc.start()
 scc.awaitTermination()
}

}

 打包运行命令：./spark-submit --class kafkaSparkStream.KafkaToSpark --master yarn-client /home/hadoop/sparkJar/kafkaToSpark.jar 192.168.1.126:9092,192.168.1.127:9092,192.168.1.128:9092 topic1

运行成功！