spark+kafka 小案例

package com.sparkstreaming

import org.apache.spark.SparkConf

import org.apache.spark.streaming.Seconds

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.streaming.kafka010.KafkaUtils

import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import org.apache.kafka.common.serialization.StringDeserializer

object SparkStreamKaflaWordCount {

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

//创建streamingContext

var conf=new SparkConf().setMaster("spark://192.168.177.120:7077")

.setAppName("SparkStreamKaflaWordCount Demo");

var ssc=new StreamingContext(conf,Seconds());

//创建topic

//var topic=Map{"test" -> 1}

var topic=Array("test");

//指定zookeeper

//创建消费者组

var group="con-consumer-group"

//消费者配置

val kafkaParam = Map(

"bootstrap.servers" -> "192.168.177.120:9092,anotherhost:9092",//用于初始化链接到集群的地址

"key.deserializer" -> classOf[StringDeserializer],

"value.deserializer" -> classOf[StringDeserializer],

//用于标识这个消费者属于哪个消费团体

"group.id" -> group,

//如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性

//可以使用这个配置，latest自动重置偏移量为最新的偏移量

"auto.offset.reset" -> "latest",

//如果是true，则这个消费者的偏移量会在后台自动提交

"enable.auto.commit" -> (false: java.lang.Boolean)

);

//创建DStream，返回接收到的输入数据

var stream=KafkaUtils.createDirectStream[String,String](ssc, PreferConsistent,Subscribe[String,String](topic,kafkaParam))

 //每一个stream都是一个ConsumerRecord

stream.map(s =>(s.key(),s.value())).print();

ssc.start();

ssc.awaitTermination();

}

}

# The number of milliseconds of each tick

tickTime=

# The number of ticks that the initial

# synchronization phase can take

initLimit=

# The number of ticks that can pass between

# sending a request and getting an acknowledgement

syncLimit=

# the directory where the snapshot is stored.

dataDir=/home/zhangxs/datainfo/developmentData/zookeeper/zkdata1

# the port at which the clients will connect

clientPort=

server.=zhangxs::

zkServer.sh start zoo1.cfg

[root@zhangxs kafka_2.]# bin/kafka-server-start.sh config/server.properties

[-- ::,] INFO KafkaConfig values:

advertised.host.name = null

advertised.listeners = null

advertised.port = null

authorizer.class.name =

auto.create.topics.enable = true

auto.leader.rebalance.enable = true

background.threads =

broker.id =

broker.id.generation.enable = true

broker.rack = null

compression.type = producer

connections.max.idle.ms =

controlled.shutdown.enable = true

controlled.shutdown.max.retries =

controlled.shutdown.retry.backoff.ms =

controller.socket.timeout.ms =

create.topic.policy.class.name = null

default.replication.factor =

delete.topic.enable = false

fetch.purgatory.purge.interval.requests =

group.max.session.timeout.ms =

group.min.session.timeout.ms =

host.name =

inter.broker.listener.name = null

inter.broker.protocol.version = 0.10.-IV0

leader.imbalance.check.interval.seconds = 

[root@zhangxs kafka_2.]# bin/kafka-console-producer.sh --broker-list 192.168.177.120: --topic test

./spark-submit --class com.sparkstreaming.SparkStreamKaflaWordCount  /usr/local/development/spark-2.0/jars/streamkafkademo.jar 

zhang xing sheng

// :: INFO cluster.CoarseGrainedSchedulerBackend$DriverEndpoint: Launching task  on executor id:  hostname: 192.168.177.120.

// :: INFO storage.BlockManagerInfo: Added broadcast_99_piece0 in memory on 192.168.177.120: (size: 1913.0 B, free: 366.3 MB)

// :: INFO scheduler.TaskSetManager: Finished task 0.0 in stage 99.0 (TID ) in  ms on 192.168.177.120 (/)

// :: INFO scheduler.TaskSchedulerImpl: Removed TaskSet 99.0, whose tasks have all completed, from pool

// :: INFO scheduler.DAGScheduler: ResultStage  (print at SparkStreamKaflaWordCount.scala:) finished in 0.019 s

// :: INFO scheduler.DAGScheduler: Job  finished: print at SparkStreamKaflaWordCount.scala:, took 0.023450 s

-------------------------------------------

Time:  ms

-------------------------------------------

(null,zhang xing sheng)

这个需要你将【spark-streaming-kafka-0-10_2.11-2.1.0】，【kafka-clients-0.10.2.0】这两个jar添加到 spark_home/jar/路径下就可以了。 （这个只是我这个工程里缺少的jar）