spark讀取kafka數據流

spark讀取kafka數據流提供了兩種方式createDstream和createDirectStream。

兩者區別如下：

1、KafkaUtils.createDstream

構造函數為KafkaUtils.createDstream(ssc, [zk], [consumer group id], [per-topic,partitions] ) 
使用了receivers來接收數據，利用的是Kafka高層次的消費者api，對于所有的receivers接收到的數據將會保存在Spark executors中，然后通過Spark Streaming啟動job來處理這些數據，默認會丟失，可啟用WAL日志，該日志存儲在HDFS上 
A、創建一個receiver來對kafka進行定時拉取數據，ssc的rdd分區和kafka的topic分區不是一個概念，故如果增加特定主體分區數僅僅是增加一個receiver中消費topic的線程數，并不增加spark的并行處理數據數量 
B、對于不同的group和topic可以使用多個receivers創建不同的DStream 
C、如果啟用了WAL，需要設置存儲級別，即KafkaUtils.createStream(….,StorageLevel.MEMORY_AND_DISK_SER)

2.KafkaUtils.createDirectStream

區別Receiver接收數據，這種方式定期地從kafka的topic+partition中查詢最新的偏移量，再根據偏移量范圍在每個batch里面處理數據，使用的是kafka的簡單消費者api 
優點: 
A、 簡化并行，不需要多個kafka輸入流，該方法將會創建和kafka分區一樣的rdd個數，而且會從kafka并行讀取。 
B、高效，這種方式并不需要WAL，WAL模式需要對數據復制兩次，第一次是被kafka復制，另一次是寫到wal中 
C、恰好一次語義(Exactly-once-semantics)，傳統的讀取kafka數據是通過kafka高層次api把偏移量寫入zookeeper中，存在數據丟失的可能性是zookeeper中和ssc的偏移量不一致。EOS通過實現kafka低層次api，偏移量僅僅被ssc保存在checkpoint中，消除了zk和ssc偏移量不一致的問題。缺點是無法使用基于zookeeper的kafka監控工具

public void adclick(){

SparkConf conf = new SparkConf()

.setAppName("")

.setMaster("");

JavaStreamingContext jssc = new JavaStreamingContext(conf,Durations.seconds(10));

jssc.checkpoint("");

Map<String,String> kafkaParams = new HashMap<String, String>();

kafkaParams.put("metadata.broker.list", ConfigurationManager.getProperty("metadata.broker.list"));

String kafkaTopics = ConfigurationManager.getProperty("kafkaTopics");

String[] kafkaTopicsSplits = kafkaTopics.split(",");

Set<String> tops = new HashSet<String>();

for(String xx:kafkaTopicsSplits){

tops.add(xx);

}

JavaPairInputDStream<String, String> adRealTimeDStream = KafkaUtils.

createDirectStream(

jssc, 

String.class, 

String.class, 

StringDecoder.class, 

StringDecoder.class, 

kafkaParams, 

tops);

jssc.start();

jssc.awaitTermination();

jssc.close();

}