前置条件：
启动kafak：
nohup kafka-server-start.sh /application/kafka_2.11-1.1.0/config/server.properties 1>/home/hadoop/logs/kafka_std.log 2>/home/hadoop/logs/kafka_err.log &创建一个没有的kafaktopic：
kafka-topics.sh --create --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181/kafka --replication-factor 3 --partitions 3 --topic zy-flume-kafka查看是否创建成功：
kafka-topics.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181/kafka --describe --topic zy-flume-kafka
配置flume的采集方案
第一级：exec-avro.conf
agent1.sources = r1
agent1.channels = c1
agent1.sinks = k1
#define sources
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /application/flume-1.8.0-bin/data/sample.log
#define channels
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 1000
agent1.channels.c1.transactionCapacity = 100
#define sink
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = hadoop02
agent1.sinks.k1.port = 3212
#bind sources and sink to channel
agent1.sources.r1.channels = c1
agent1.sinks.k1.channel = c1第二级：avro-kafka.conf
agent2.sources = r2
agent2.channels = c2
agent2.sinks = k2
#define sources
agent2.sources.r2.type = avro
agent2.sources.r2.bind = hadoop02
agent2.sources.r2.port = 3212
#define channels
agent2.channels.c2.type = memory
agent2.channels.c2.capacity = 1000
agent2.channels.c2.transactionCapacity = 100
#define sink
agent2.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink
agent2.sinks.k2.brokerList = hadoop01:9092,hadoop02:9092,hadoop03:9092
agent2.sinks.k2.topic = zy-flume-kafka
agent2.sinks.k2.batchSize = 4
agent2.sinks.k2.requiredAcks = 1
#bind sources and sink to channel
agent2.sources.r2.channels = c2
agent2.sinks.k2.channel = c2启动flume
hadoop02：
flume-ng agent --conf /application/flume-1.8.0-bin/conf/ --name agent2 --conf-file /application/flume-1.8.0-bin/flume_sh/avro-kafka.conf -Dflume.root.logger=DEBUG,consolehadoop01：
flume-ng agent --conf /application/flume-1.8.0-bin/conf/ --name agent1 --conf-file /application/flume-1.8.0-bin/flume_sh/exec-avro.conf -Dflume.root.logger=DEBUG,console注意：一定要先启动第二级在启动第一级
测试
启动一个kafakconsumer
kafka-console-consumer.sh --bootstrap-server hadoop01:9092,hadoop02:9092,hadoop03:9092 --from-beginning --topic zy-flume-kafka向监控文件下添加数据：tail -10 sample.temp>>sample.log
观察kafkaconsumer：消费到数据！！
SparkStreaming整合kafka有两种方式：
- receiver +checkpoint方式
- direct +zookeeper方式
代码：
/**
* 基于Receiver的方式去读取kafka中的数据
*/
object _01SparkKafkaReceiverOps {
def main(args: Array[String]): Unit = {
//判断程序传入的参数个数是否正确
//2 hadoop01:2181,hadoop02:2181，hadoop03:2181/kafka first zy-flume-kafka
if (args == null || args.length < 4) {
println(
"""
|Parameter Errors! Usage: <batchInterval> <zkQuorum> <groupId> <topics>
|batchInterval : 批次间隔时间
|zkQuorum : zookeeper url地址
|groupId : 消费组的id
|topic : 读取的topic
""".stripMargin)
System.exit(-1)
}
//获取程序传入的参数
val Array(batchInterval, zkQuorum, groupId, topic) = args
//1.构建程序入口
val conf: SparkConf = new SparkConf()
.setMaster("local[2]")
.setAppName("_01SparkKafkaReceiverOps")
val ssc =new StreamingContext(conf,Seconds(2))
/**2.使用Receiver方式读取数据
* @param ssc
* @param zkQuorum
* @param groupId
* @param topics
* @param storageLevel default: StorageLevel.MEMORY_AND_DISK_SER_2
* @return DStream of (Kafka message key, Kafka message value)
*/
val topics = topic.split("\\s+").map((_,3)).toMap
//2.读取数据
val message: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc,zkQuorum,groupId,topics)
//3.打印数据
message.print()
//4.提交任务
ssc.start()
ssc.awaitTermination()
}
}注意（receiver +checkpoint）：
- kafka中的topic和sparkstreaming中生成的RDD分区没有关系，在KafkaUtils.createStream中增加分区数只会增加单个receiver的线程数，不会增加spark的并行度
- 可以创建多个kafka的输入DStream，使用不同的group和topic，使用多个receiver并行接收数据
- 如果启用了HDFS等有容错的存储系统，并且启用了写入日，则接收到的数据已经被复制到日志中。
代码实现
package com.zy.streaming
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.curator.framework.{CuratorFramework, CuratorFrameworkFactory}
import org.apache.curator.retry.ExponentialBackoffRetry
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* 使用zk来管理的消费的偏移量，确保当SparkStreaming挂掉之后在重启的时候，
* 能够从正确的offset偏移量的位置开始消费，而不是从头开始消费
*/
object SparkStreamingDriverHAOps {
//设置zookeeper中存放偏移量的位置
val zkTopicOffsetPath="/offset"
//获取zookeeper的编程入口
val client:CuratorFramework={
val client=CuratorFrameworkFactory.builder()
.connectString("hadoop01:2181,hadoop02:2181,hadoop03:2181/kafka")
.namespace("2019_1_7")
.retryPolicy(new ExponentialBackoffRetry(1000,3))
.build()
client.start()
client
}
def main(args: Array[String]): Unit = {
//屏蔽日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//2 direct zy-flume-kafka
if(args==null||args.length<3){
println(
"""
|Parameter Errors! Usage: <batchInterval> <groupId> <topics>
|batchInterval : 批次间隔时间
|groupId : 消费组的id
|topic : 读取的topic
""".stripMargin)
System.exit(-1)
}
//获取传入的参数
val Array(batchInterval,groupId,topic)=args
//1.构建程序入口
val conf: SparkConf = new SparkConf()
.setMaster("local[2]")
.setAppName("SparkStreamingDriverHAOps")
val ssc =new StreamingContext(conf,Seconds(batchInterval.toLong))
//连接kafka的参数
val kafkaParams=Map(
"bootstrap.servers"->"hadoop01:9092,hadoop02:9092,hadoop03:9092", //集群入口
"auto.offset.reset"->"smallest" //消费方式
)
//2.创建kafka的message
val message:DStream[(String,String)]=createMessage(topic,groupId,ssc,kafkaParams)
//3.业务处理，这里主要是介绍如何kafka整合sparkStreaming，所以这里不做业务处理
message.foreachRDD(rdd=>{
if(!rdd.isEmpty()){
println(
"""
|####################>_<####################
""".stripMargin+rdd.count())
}
//更新偏移量
storeOffsets(rdd.asInstanceOf[HasOffsetRanges].offsetRanges,groupId)
})
//4.启动程序
ssc.start()
ssc.awaitTermination()
}
/**
* 创建kafka对应的message
* 分两种情况：
* 1.第一次消费的时候，从zk中读取不到偏移量
* 2.之后的消费从zk中才能读取到偏移量
*/
def createMessage(topic: String, groupId: String, ssc: StreamingContext, kafkaParams: Map[String, String]): InputDStream[(String, String)] = {
//获取偏移量，以及判断是否是第一次消费
val (fromOffsets,flag)=getFromOffsets(topic, groupId)
var message:InputDStream[(String, String)] = null
//构建kafka对应的message
if(flag){ //标记位，使用zk中得到的对应的partition偏移量信息，如果有为true
/**
* recordClass: Class[R],
* kafkaParams: JMap[String, String],
* fromOffsets: JMap[TopicAndPartition, JLong],
* messageHandler: JFunction[MessageAndMetadata[K, V], R]
*/
val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key, mmd.message)
message = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder,(String,String)](ssc,kafkaParams,fromOffsets,messageHandler)
}else{ //如果是第一次读取为false
/**
* createDirectStream[
* String, key的类型
* String, value的类型
* StringDecoder, key的序列化的类型
* StringDecoder] value的序列化的类型
*
*/
message=KafkaUtils.createDirectStream[String,
String,
StringDecoder
, StringDecoder](ssc,kafkaParams,topic.split("\\s+").toSet)
}
message
}
//获取对应的topic中的每一个partition的偏移量信息
def getFromOffsets(topic: String, groupId: String):(Map[TopicAndPartition, Long], Boolean)= {
//构建存储offset的位置信息的路径
val zkPath=s"${zkTopicOffsetPath}/${topic}/${groupId}"
//判断当前路径是否存在，不存在则创建
nsureZKPathExists(zkPath)
//获取所有分区中存储的offset信息
import scala.collection.JavaConversions._
val offsets=for{p<-client.getChildren.forPath(zkPath)}yield{
val offset=new String(client.getData.forPath(s"${zkPath}/${p}")).toLong
(new TopicAndPartition(topic,p.toInt),offset)
}
//如果未空表示第一次读取,无偏移量信息
if(offsets.isEmpty){
(offsets.toMap,false)
}else{
(offsets.toMap,true)
}
}
def storeOffsets(offsetRanges: Array[OffsetRange], groupId: String): Unit = {
for(offsetRange<-offsetRanges){
val partition=offsetRange.partition
val topic=offsetRange.topic
//获取偏移量
val offset=offsetRange.untilOffset
//构建存放偏移量的znode
val path=s"${zkTopicOffsetPath}/${topic}/${groupId}/${partition}"
//判断是否存在，不存在则创建
nsureZKPathExists(path)
client.setData().forPath(path,(""+offset).getBytes())
}
}
def nsureZKPathExists(zkPath: String) = {
//如果为空的话就创建
if(client.checkExists().forPath(zkPath)==null){
//如果父目录不存在，连父目录一起创建
client.create().creatingParentsIfNeeded().forPath(zkPath)
}
}
}注意（direct +zookeeper）：
- 不需要创建多个输入kafka流并将其合并，使用directStream，spark Streaming将创建于使用kafka分区一样多的RDD分区，这些分区的数据全部从kafka并行读取数据，kafka和RDD分区之间有一对一的映射关系。
- Direct方式没有接收器，不需要预先写入日志，只要kafka数据保留时间足够长就行
- 保证了正好一次的消费语义（offset保存在zookeeper中）
原文：http://blog.51cto.com/14048416/2339917