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(4));
//创建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=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
dataDir=/home/zhangxs/datainfo/developmentData/zookeeper/zkdata1
# the port at which the clients will connect
clientPort=2181
server.1=zhangxs:2881:3881

zkServer.sh start zoo1.cfg

[root@zhangxs kafka_2.11]# bin/kafka-server-start.sh config/server.properties
[2017-03-25 18:42:03,153] 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 = 10
broker.id = 0
broker.id.generation.enable = true
broker.rack = null
compression.type = producer
connections.max.idle.ms = 600000
controlled.shutdown.enable = true
controlled.shutdown.max.retries = 3
controlled.shutdown.retry.backoff.ms = 5000
controller.socket.timeout.ms = 30000
create.topic.policy.class.name = null
default.replication.factor = 1
delete.topic.enable = false
fetch.purgatory.purge.interval.requests = 1000
group.max.session.timeout.ms = 300000
group.min.session.timeout.ms = 6000
host.name =
inter.broker.listener.name = null
inter.broker.protocol.version = 0.10.2-IV0
leader.imbalance.check.interval.seconds = 300

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

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

zhang xing sheng