本篇文章給大家分享的是有關(guān)Flume+Kafka+SparkStreaming的整合是怎么樣的，小編覺得挺實用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

成都創(chuàng)新互聯(lián)專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務(wù)，包含不限于網(wǎng)站設(shè)計、成都網(wǎng)站建設(shè)、西寧網(wǎng)絡(luò)推廣、重慶小程序開發(fā)、西寧網(wǎng)絡(luò)營銷、西寧企業(yè)策劃、西寧品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運營等，從售前售中售后，我們都將竭誠為您服務(wù)，您的肯定，是我們最大的嘉獎；成都創(chuàng)新互聯(lián)為所有大學(xué)生創(chuàng)業(yè)者提供西寧建站搭建服務(wù)，24小時服務(wù)熱線：13518219792，官方網(wǎng)址：aaarwkj.com

1.架構(gòu)

第一步，F(xiàn)lume和Kakfa對接，F(xiàn)lume抓取日志，寫到Kafka中

第二部，Spark Streaming讀取Kafka中的數(shù)據(jù)，進行實時分析

下面首先使用Kakfa自帶的消息處理（腳本）來獲取消息，走通Flume和Kafka的對接

2.安裝flume,kafka

flume install: http://my.oschina.net/u/192561/blog/692225

kafka install: http://my.oschina.net/u/192561/blog/692357

3.Flume和Kafka整合

3.1 兩者整合優(yōu)勢

Flume更傾向于數(shù)據(jù)傳輸本身，Kakfa是典型的消息中間件用于解耦生產(chǎn)者消費者。

具體架構(gòu)上，Agent并沒把數(shù)據(jù)直接發(fā)送到Kafka，在Kafka前面有層由Flume構(gòu)成的forward。這樣做有兩個原因：

Kafka的API對非JVM系的語言支持很不友好，forward對外提供更加通用的HTTP接口。forward層可以做路由、Kafka topic和Kafkapartition key等邏輯，進一步減少Agent端的邏輯。

數(shù)據(jù)有數(shù)據(jù)源到flume再到Kafka時，數(shù)據(jù)一方面可以同步到HDFS做離線計算，另一方面可以做實時計算。本文實時計算采用SparkStreaming做測試。
 

3.2 Flume和Kafka整合安裝

1. 下載Flume和Kafka集成的插件，下載地址：

https://github.com/beyondj2ee/flumeng-kafka- plugin

將package目錄中的flumeng-kafka-plugin.jar拷貝到Flume安裝目錄的lib目錄下

2. 將Kakfa安裝目錄libs目錄下的如下jar包拷貝到Flume安裝目錄的lib目錄下

kafka_2.11-0.10.0.0.jar

scala-library-2.11.8.jar

metrics-core-2.2.0.jar

提取插件中的flume-conf.properties文件：修改如下:flume源采用exec

producer.sources.s.type = exec
producer.sources.s.command=tail -F -n+1 /home/eric/bigdata/kafka-logs/a.log
producer.sources.s.channels = c1

修改producer代理的topic為 HappyBirthDayToAnYuan

將配置放到 apache-flume-1.6.0-bin/conf/producer.conf中

 完整 producer.conf:   

#agentsectionproducer.sources= s1producer.channels= c1producer.sinks= k1#配置數(shù)據(jù)源producer.sources.s1.type=exec#配置需要監(jiān)控的日志輸出文件或目錄producer.sources.s1.command=tail -F -n+1 /home/eric/bigdata/kafka-logs/a.log#配置數(shù)據(jù)通道producer.channels.c1.type=memoryproducer.channels.c1.capacity=10000producer.channels.c1.transactionCapacity=100#配置數(shù)據(jù)源輸出#設(shè)置Kafka接收器,此處最坑,注意版本,此處為Flume 1.6.0的輸出槽類型producer.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink#設(shè)置Kafka的broker地址和端口號producer.sinks.k1.brokerList=localhost:9092#設(shè)置Kafka的Topicproducer.sinks.k1.topic=HappyBirthDayToAnYuan#設(shè)置序列化方式producer.sinks.k1.serializer.class=kafka.serializer.StringEncoder#將三者級聯(lián)producer.sources.s1.channels=c1producer.sinks.k1.channel=c1

3.3 啟動kafka flume相關(guān)服務(wù)

啟動ZK bin/zookeeper-server-start.sh config/zookeeper.properties

啟動Kafka服務(wù)bin/kafka-server-start.sh config/server.properties

創(chuàng)建主題

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic HappyBirthDayToAnYuan

查看主題

bin/kafka-topics.sh --list --zookeeper localhost:2181

查看主題詳情

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic HappyBirthDayToAnYuan

刪除主題

bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic test

創(chuàng)建消費者

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

啟動flume

bin/flume-ng agent -n producer -c conf -f conf/producer.conf -Dflume.root.logger=INFO,console

向flume發(fā)送數(shù)據(jù):

echo "yuhai" >> a.log

kafka消費數(shù)據(jù):

注意:當(dāng)前文件內(nèi)容刪除,服務(wù)器重啟,主題需重新創(chuàng)建,但是消費內(nèi)容有落地文件,當(dāng)前消費內(nèi)容不消失.

以上就是Flume+Kafka+SparkStreaming的整合是怎么樣的，小編相信有部分知識點可能是我們?nèi)粘９ぷ鲿姷交蛴玫降?。希望你能通過這篇文章學(xué)到更多知識。更多詳情敬請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

分享名稱：Flume+Kafka+SparkStreaming的整合是怎么樣的
瀏覽地址：http://aaarwkj.com/article38/jjjpsp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供建站公司、手機網(wǎng)站建設(shè)、網(wǎng)站改版、電子商務(wù)、網(wǎng)站內(nèi)鏈、靜態(tài)網(wǎng)站

廣告

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