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();

}

本文標題：spark讀取kafka數據流
當前鏈接：http://vcdvsql.cn/article14/gjepge.html

成都網站建設公司_創新互聯，為您提供App開發、軟件開發、Google、外貿網站建設、網站營銷、企業建站

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯