內(nèi)容課件講稿flume-day

1、一課前提1. flume簡(jiǎn)一課前提1. flume簡(jiǎn)selector相關(guān)配置與案例分Sink ProAVROnt*開(kāi)發(fā)*(選講10.flume和kafka的整合(后期講2. 二正課內(nèi)第一節(jié)：Flume1.1 大數(shù)據(jù)處理流在企業(yè)中，大數(shù)據(jù)的處理流程一般是-數(shù)數(shù)數(shù)據(jù)分?jǐn)?shù)據(jù)展參考下圖在和搜集的工具中，F(xiàn)lume框架占有一定的市場(chǎng)份量1.2 Flume的簡(jiǎn)Flume是一種分布式的，可靠的、高可用的服務(wù)，用于有效地收集，聚合和移動(dòng)大量日志數(shù)據(jù)。它具有強(qiáng)大的功能和容錯(cuò)能力。它使用一個(gè)簡(jiǎn)單的可擴(kuò)展數(shù)據(jù)模型，允分析在和搜集的工具中，F(xiàn)lume框架占有一定的市場(chǎng)份量1.2 Flume的簡(jiǎn)Flume是一種分布式的

2、，可靠的、高可用的服務(wù)，用于有效地收集，聚合和移動(dòng)大量日志數(shù)據(jù)。它具有強(qiáng)大的功能和容錯(cuò)能力。它使用一個(gè)簡(jiǎn)單的可擴(kuò)展數(shù)據(jù)模型，允分析應(yīng)用程序參考官網(wǎng)： HYPERLINK http:/ume.apache/ /Flume is a distributed, reliable, and available service for eciently collecting, andmovinglargeamountsoflogdata.s a simple and exible architecture based streamingdataows.Itisrobustandfaulttolerant

3、withtunablereliabilitymechanismsmany failover and recovery mechanisms. It uses a simple extensible data tforyticume 最開(kāi)始是由 cloudera 開(kāi)發(fā)的實(shí)時(shí)日志收集系統(tǒng)，受到了業(yè)界的認(rèn)可與廣泛應(yīng)用。但隨著 功能的擴(kuò)展，ume的代碼工程臃腫組件設(shè)計(jì)不合理配置不標(biāo)準(zhǔn)等缺點(diǎn)漸出來(lái)，其是版本 志傳輸不穩(wěn)定的現(xiàn)象尤為嚴(yán)重為了解決這些問(wèn)題，2011 年 10 月 22 號(hào)，cloudera 對(duì) Flume 進(jìn)行了里程碑式的改動(dòng)：重組件Flume配置以及代碼架構(gòu)，并將 Flume 納入 apa

4、che 旗下，從cloudera Flume 改名為 1.3 版本區(qū)為了與之前版本區(qū)分開(kāi)，重構(gòu)后的版本統(tǒng)稱為 Flume NG（next generation），重構(gòu)前的版本被統(tǒng)FlumeOG（originalgeneration），F(xiàn)lume目前只有Linux系統(tǒng)的啟，沒(méi)有Windows環(huán)境啟。第二節(jié)：Flume2.1 體系結(jié)構(gòu)簡(jiǎn)Flume運(yùn)行是 Agent。Flume是以agent為最小的獨(dú)立運(yùn)。一個(gè)agent就是一個(gè)JVM。它一個(gè)完整的數(shù)據(jù)收集工具，含有三組件，分別是source、 channel、 sink。通過(guò)這些組件Event 可以從一個(gè)地方流向另一個(gè)地方。如下圖所示2.2 組件及

5、其作客戶端-nt生產(chǎn)數(shù)據(jù)，運(yùn)行在一個(gè)獨(dú)立的線程-一個(gè)數(shù)據(jù)單元，消息頭2.2 組件及其作客戶端-nt生產(chǎn)數(shù)據(jù)，運(yùn)行在一個(gè)獨(dú)立的線程-一個(gè)數(shù)據(jù)單元，消息頭和消息體組成。（Events可以是日、 avro對(duì)象等。Event從源點(diǎn)到達(dá)目的點(diǎn)的遷移的抽象-一個(gè)獨(dú)立的Flume進(jìn)程，運(yùn)行在JVM中，包含組件Source、 Channel、 Sink每臺(tái)機(jī)器運(yùn)行一個(gè)agent，但是一個(gè)agent中可以包含多個(gè)和sinks-數(shù)據(jù)收集組件。source從nt收集數(shù)據(jù)，傳遞給管道，負(fù)責(zé)接收source端的數(shù)據(jù)，然后將數(shù)據(jù)推送到sink端-負(fù)責(zé)從取數(shù)據(jù)，并將其推送到持久化系統(tǒng)或者是下一個(gè)Agent選擇器，作用于so

6、urce端，然后決定數(shù)據(jù)發(fā)往哪個(gè)目標(biāo)-器，flume允許使器數(shù)據(jù)。允許使器鏈，作用于source和sink階段第三節(jié)：Flume3.1 安裝和配置環(huán)境變3.1.1 準(zhǔn)包將apache-flume-1.8.0-ar.gz 上傳到linux系統(tǒng)中的/opt/software/目錄3.1.2 解包rootmastertar -zxvf apache-flume-1.8.0-ar.gz-C3.1.3 更名操rootmastersoftware#cdrootmasterapps#mvapache-flume-1.8.0-bin/3.1.4 配置環(huán)境變量，并rootmaster3.1.3 更名操rootm

7、astersoftware#cdrootmasterapps#mvapache-flume-1.8.0-bin/3.1.4 配置環(huán)境變量，并rootmasterapps#vi.省略export FLUME_HOME=/opt/apps/flume exportrootmasterapps#source3.1.5 驗(yàn)證環(huán)境變r(jià)ootmasterapps#flume-ngFlumeSourcecode: CompiledbydenesonFriSep15TFromsourcewithchecksum3.2 配置文flume-env.sh.template flume-env.shflume-.省略

8、export.省略第四節(jié)：Flume4.1 數(shù)據(jù)模單一數(shù)據(jù)模多數(shù)據(jù)流模4.1.1 單一數(shù)據(jù)模在單個(gè) Agent 內(nèi)由單個(gè) Source, Channel, Sink 建立一個(gè)單一的數(shù)據(jù)流模型，如下圖所示，整個(gè)數(shù)據(jù)流WebServerSourceChannelSinkHDFS。4.1.2 多數(shù)據(jù)流4.1.2 多數(shù)據(jù)流模1）多 Agent 串行傳輸數(shù)據(jù)流模2）多 Agent 匯聚數(shù)據(jù)流模3）單 Agent 多路數(shù)據(jù)流模4）Sinkgroups 數(shù)據(jù)流模4.1.3 小總在flume提供的數(shù)據(jù)流模型中，幾個(gè)原則4）Sinkgroups 數(shù)據(jù)流模4.1.3 小總在flume提供的數(shù)據(jù)流模型中，幾個(gè)原則很

9、重要Source-1.單個(gè)Source組件可以和多個(gè)Channel組合建立數(shù)據(jù)流，既可以multiplexing2.多個(gè)可以寫(xiě)入單個(gè)Channel-Sinkgroups從Channel中獲取數(shù)據(jù)，既可以loadbalancing和failover多個(gè)Sinks也可以從單個(gè)Channel中取數(shù)據(jù)單個(gè)Sink只能從單個(gè)Channel中取數(shù)根據(jù)上述 5 個(gè)原則，你可以設(shè)計(jì)出滿足你需求的數(shù)據(jù)流模型配置介定義組件名然后將源道。格式如下中的流，您需要通過(guò)通源。您需要列出給的源和通道指向一個(gè)通道。一個(gè)源實(shí)例可以指定多個(gè)通道，但是一實(shí)例只能指定一個(gè)#listthe.sinks ,sinksandchannel

10、s=.channels=#listthe.sinks ,sinksandchannels=.channels=#setfor.channels=#setfor.sinks.channel = 案例如下#listthe,sinksandchannels= avro-appserver-src-=hdfs-sink-agent_foo.channels = mem-channel-#setchannelfor.avro-appserver-src-1.channels = mem-channel-#setchannelforagent_foo.sinks.hdfs-sink-1.channel =

11、 mem-channel-4.2.2 配置組件屬#for. = #for.channel. = #propertiesfor. = 案例如下agent_foo.sinks = hdfs-Cluster1-sink agent_foo.channels = mem-channel-1#setchannelfor,# properties of avro-AppSrv-source .avro-AppSrv-source.port = 10000# properties of mem-channel-1 agent_foo.channels.mem-channel-1.type = memory

12、agent_foo.channels.mem-channel-1.capacity = 1000agent_foo.channels.mem-channel-ionCapacity=# properties of hdfs-Cluster1-sink agent_foo.sinks.hdfs-Cluster1-sink.type = hdfs4.3 常用的source和sink種參考flume文4.3.1常用的ume4.3 常用的source和sink種參考flume文4.3.1常用的umeSyslogTCPsource： SyslogUDPSource： JMSsource： Spoolin

13、gdirectorysource： Kafka#4.3.2常用的ume#JDBCChannel KafkaFile Channel#4.3.3常用的umeagent_foo.sinks.hdfs-Cluster1-sink.hdfs.path = #HDFSSink HIVESink LoggerSink AvroSink KafkaHbase Sink#5.1 案例演示#HDFSSink HIVESink LoggerSink AvroSink KafkaHbase Sink#5.1 案例演示5.1.1 編方flume# mkdir flume# cd flumeconf# vi avro-

14、#定義各個(gè)組件的名=avro-a1.channels=mem-a1.sinks=logger-組件的相關(guān)屬.avro-.avro-.avro-#定義channels組件的相關(guān)屬a1.channels.mem-a1.sinks.logger-#組件之間進(jìn)行綁.avro-sour1.channels=mem-a1.sinks.logger-sink1.channel=mem-5.1.2 啟動(dòng)rootmasterflumeconf#flume-ngagent-c-f./avro-logger.conf-na15.1.3 測(cè)試數(shù)#mkdir#cdflumedata/ flumedata#dateflu

15、medata#cat2019年 11月 21四 21:22:36.省略#mkdir#cdflumedata/ flumedata#dateflumedata#cat2019年 11月 21四 21:22:36.省略mastercatflume-ng avro-nt-c/usr/local/flume-master-p9999-F實(shí)文件配置方rootmaster flumeconf# vi exec-mand=tail-F 5.2.2 啟動(dòng)rootmasterflumeconf#flume-ng-c./conf-f./exec-na15.2.3 測(cè)試數(shù)rootmastermaster5.3 實(shí)(

16、文件5.3.1 配置方rootmaster flumeconf# vi exec-=a2.channels=a2.sinks=5.3.1 配置方rootmaster flumeconf# vi exec-=a2.channels=a2.sinks=.r1.type=mand=tail-Fa2.channels.c1.keep-5.3.2 啟動(dòng)rootmaster flumeconf# flume-ng agent -c -f./exec-logger.conf-n5.3.3 測(cè)試rootmaster # echo nice 實(shí)配置方目錄)：spool +le + rootmaster flum

17、econf# vi spool-hdfs.conf = r1a1.channels=c1 a1.sinks = k1.r1.type=.r1.spoolDir=a1.channels.c1.type=Dir=/home/flume/checkpo a1.channels.c1.dataDirs = /home/flume/dataa1.sinks.k1.type=a1.sinks.k1.hdfs.path = hdfs:/master:8020/flume/spooldir a1.sinks.k1.hdfs.filePrefix =a1.sinks.k1.hdfs.round = true a

18、1.sinks.k1.hdfs.roundUnit = a1.sinks.k1.hdfs.fileSuffix= .r1.channels=a1.sinks.k1.channela1.sinks.k1.hdfs.roundUnit = a1.sinks.k1.hdfs.fileSuffix= .r1.channels=a1.sinks.k1.channel=5.4.2 啟動(dòng)rootmaster flumeconf# flume-ng agent -c ./conf/ -f ./spool-hdfs.conf -n -5.4.3 測(cè)-1. 向/home/flume/input/2020/01/目

19、錄里添加幾個(gè)文#cdo world a1.log zuonodiea2.logligei-2. 查看hdfs的目錄內(nèi)容-3. 在查看一下home/flume/input/2020/01/目錄里的內(nèi)rootmaster01#實(shí)配置方目錄)：spool+ rootmaster flumeconf# vi spool-logger.conf = r1a1.channels=c1 a1.sinks = s1.r1.fileSuffix = .COMPLETED 5.5.2 啟動(dòng)rootmasterflumeconf#5.5.2 啟動(dòng)rootmasterflumeconf#flume-ngagent-c

20、./conf-f./spool-logger.conf-n-5.5.3 測(cè)試rootmaster#foriinseq110;doecho$itaildirSource源ume1.7.0加入了taildirSource作為agent的source?？梢哉f(shuō)是spooling directory 的結(jié)合體?？梢钥蓛?yōu)點(diǎn)能支持?jǐn)帱c(diǎn)續(xù)傳多個(gè)目錄，并且使用正則表達(dá)式匹配該目錄中的文件名進(jìn)行實(shí)時(shí)收集可多個(gè)文件也是近實(shí)，需要配置一個(gè)位置文件減少數(shù)據(jù)丟升級(jí)roo doop01local#mv/usr/local/flume-1.6.0/usr/local/flume-1.6.0_bak roo doop02loc

21、al#mv/usr/local/flume-1.6.0/usr/local/flume-1.6.0_bak roo doop02local#mv/usr/local/flume-1.6.0/usr/local/flume-doop01local#tar-zxvf/home/apache-flume-1.7.0-ar.gz-doop01 local# vi /etc/profile doop01local#source/etc/profiledoop01local#scp-r./flume-1.7.0/hadoop02:/usr/local/ doop01local#scp-r./flume-1

22、.7.0/roo doop01local#scp-r/etc/profilehadoop02:/etc/ roo doop01local#scp-r/etc/profilehadoop03:/etc/ roo doop01 local# flume-ng vervi創(chuàng)目錄flume-flume-mkdir-mkdir-測(cè)試vi創(chuàng)目錄flume-flume-mkdir-mkdir-測(cè)試flume-flume-agent-c./conf-f./conf/ts.conf-na1測(cè)試斷點(diǎn)續(xù)傳當(dāng)ume由于未知錯(cuò)誤停止運(yùn)行后，日志仍然在不斷增長(zhǎng)，此時(shí)手動(dòng)重新啟動(dòng)ume，ume可以根據(jù)錄點(diǎn)接續(xù)上次位置繼。d

23、oop01flume-1.7.0#flume-ngagent-c./conf-f./conf/ts.conf-na1taildirSource問(wèn)題# source的名字 = r1 a1.channels=c1 a1.sinks = s1#source配.r1.type=itionFile = /home/flume/taildir_.r1.filegroups = f1 f2 .r1.headers.f1.headerKey1 = value1.r1.headers.f2.headerKey1 = value2.r1.fileHeader = trueionCapacity= a1.sinks

24、.s1.typelogger#綁.r1.channels=c1 a1.sinks.s1.channel = c11、log4j的日志文件肯定是會(huì)根據(jù)規(guī)則進(jìn)行滾動(dòng)的：當(dāng)*.log滿了就會(huì)滾動(dòng)把前文件更名為*.log.1,然后新進(jìn)行*.log文件打印。這樣flume就會(huì)1、log4j的日志文件肯定是會(huì)根據(jù)規(guī)則進(jìn)行滾動(dòng)的：當(dāng)*.log滿了就會(huì)滾動(dòng)把前文件更名為*.log.1,然后新進(jìn)行*.log文件打印。這樣flume就會(huì)把*.log.1文件當(dāng)作新文件，又重一遍，導(dǎo)致重復(fù)2、當(dāng)?shù)娜罩疚募灰谱呋騽h除，flume仍然中，并沒(méi)資源，當(dāng)然，在一定時(shí)間后自，這個(gè)時(shí)間根文檔設(shè)置默認(rèn)值是120000ms問(wèn)題:u

25、me會(huì)把重命名的文件重新當(dāng)作新文是因?yàn)檎齽t表達(dá)式的原因，因?yàn)橹孛蟮奈募匀缓险齽t表達(dá)式。所以第一，重命名后的文件仍然會(huì)被；第二，ume是根據(jù)文件inode&件絕對(duì)路徑 、文件是否為null&文件絕對(duì)路徑，這樣的條件來(lái)判斷是否是同一個(gè)文件這個(gè)可以看碼源碼，放到maven項(xiàng)目（注意路徑名稱對(duì)應(yīng)），找到taildirsource的包 publicListskipToEnd)throwsIOExceptionupdateTime = System.currentTimeMillis(); for (TaildirMatcher taildir : taildirCache) Map header

26、s = headerTable.row(taildir.getFileGroup(); for (File f : taildir.getMatchingFiles() longinode=TailFiletf=if (tf = null | !tf.getPath().equals(f.getAbsolutePath() 改if (tf = null)tf=skipToEnd?f.length():openFile(f,headers,inode,updated=tf.getLastUpdated()if(updated)if(tf.getRaf()=null)tf=openFile(f,h

27、eaders,inode,if(f.length()()+()+nfilesize!+Restarting0,file:tf.getPath()+,inode:+(tf.getPath(),inode,returnTailFile類的方法(Stringpath,longinode,)throws/需要修改為if(this.inode=inode&this.path.equals(path)(this.inode(Stringpath,longinode,)throws/需要修改為if(this.inode=inode&this.path.equals(path)(this.inode=(iti

28、on,file:+path+,inode:+inode+:returnreturn將2個(gè)文件修改之后重新打包，然后放到ume的lib目錄下問(wèn)題：如果這個(gè)文件在默認(rèn)值120000ms內(nèi)都沒(méi)有新行append，就會(huì)關(guān)閉資源append就自動(dòng)打開(kāi)該資源改小該值：會(huì)造成頻繁打開(kāi)資源，也會(huì)耗時(shí)改大該值：不會(huì)頻繁打開(kāi)文件，但是已經(jīng)的資源將不會(huì)在短時(shí)資源；而當(dāng)有新。案例演示：http+配置方rootmaster flumeconf# vi http-=a1.channels=.r1.bind=.r1.port=.r1.handler .r1.handler.nickname=random5.6.2 啟動(dòng)ag

29、ent的服務(wù)rootmaster#flume-ngagent-c./conf-f./http-logger.conf-5.6.3 測(cè)試rootmaster#curl-T-d5.6.2 啟動(dòng)agent的服務(wù)rootmaster#flume-ngagent-c./conf-f./http-logger.conf-5.6.3 測(cè)試rootmaster#curl-T-dname:zhangsan,pwd:123456,body:this 案例演示配置方rootmaster flumeconf# vi syslogtcp-=a1.channels=.r1.host=.r1.port=5.7.2 啟動(dòng)ro

30、otmasterflumeconf#flume-ngagent-ca1-./syslogtcp-5.7.3 測(cè)試：需要先安裝rootmaster#yum-yinstallnmap-rootmaster#echooworld|ncmaster案例演示配置方rootmaster flumeconf# vi syslogtcp-=a1.channels=.r1.host=.r1.port=.r1.host=.r1.port= 5.8.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-nga1-c-f ./syslogtcp-5.8.3 測(cè)試rootmaster#echooo

31、|ncmaster案例演示配置方rootmaster flumeconf# vi syslogtcp-=a1.channels=.r1.host=.r1.port=a1.channels.c1.keep- a1.channels.c1.keep- 5.9.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-ng-c./conf-f./syslogtcp-n5.9.3 測(cè)試rootmaster#echoooqiangeng|ncmaster6.1 常器S案例演示配置方案rootmaster flumeconf# vi ts.conf = r1a1.channels=c1

32、a1.sinks = s1.r1.host=.r1.port=erceptors=i1 i2 i3 .r1.host=.r1.port=erceptors=i1 i2 i3 6.2.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-ng-c./conf-f-na16.2.3 測(cè)試rootmaster#echoooqiangeng|ncmaster案例演示配置方rootmaster flumeconf# vi regex-ts.conf = r16.3.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-ngagent-c-f6.3.2 啟動(dòng)ag

33、ent的服務(wù)rootmasterflumeconf#flume-ngagent-c-f./regex-ts.conf-na16.3.3 測(cè)試rootmaster#echooo|ncmasterqiangeng|ncmasterrootmaster#echooo6.4 正器via1.channels=c1 a1.sinks = s1.r1.host = master .r1.port = 6666 #定義三個(gè)插=a1.channels=c1 a1.sinks = s1#配置source插=erceptors = i1 i2 erceptors.i1.type = serceptors.i1.ke

34、y = city erceptors.i1.value = beijing erceptors.i2.type=erceptors.i2.regex=0-=#配置#定義三個(gè)插=a1.channels=c1 a1.sinks = s1#配置source插=erceptors = i1 i2 erceptors.i1.type = serceptors.i1.key = city erceptors.i1.value = beijing erceptors.i2.type=erceptors.i2.regex=0-=#配置a1.channels.c1.type = memory ionCapaci

35、ty=a1.channels.c1.byteCapacity = #配置a1.sinks.s1.type=#連接三個(gè)插.r1.channels=a1.sinks.s1.channel=測(cè)試doop01flume-1.6.0#flume-ng-c-測(cè)試數(shù)據(jù)doop01#curl-T-dk1:param1,k2:parm2,body:666 is doop01#curl-T-dk1:param1,k2:parm2,body:this is 6.5 自定6.4.0 需求器對(duì)每一組的item_type和active_time都過(guò)濾出相應(yīng)的HOST和6.4.1 flume-ng-flume-ng-6.4

36、.2 代public class Myerceptor implementserceptor publicvoidinitialize()publicvoidclose()單條parampublicercept(Eventevent)/輸String/輸bytetry inputeBody=new String(event.getBody(), Charsets.UTF_8); ArrayList temp = new ArrayList();JSONObjectbodyObj=/1)String host = bodyObj.getString(host); Stringuser_id=bo

37、dyObj.getString(user_id); /2)Json數(shù)組=everyjsonobj for (Object item : data) JSONObject itemObj = JSON.parseObject(item.toString(); HashMap fields = new HashMap(); temp.add(new /3)Jsonobj拼catch(Exceptionln(輸入數(shù)據(jù)return event;一批parampublicercept(List events) /輸出-一批ArrayListresult=newfor(Eventevent:events)

38、/一條erceptedEvent erceptedEventBody StringmultiEvent=newfor (String needEvent : multiEventArr) SimpleEventsimpleEvent=newSimpleEvent(); catch(Exceptionreturn*實(shí)publics類接icclassBuilder6.4.3 打包上使用maven器打6.4.3 打包上使用maven器打包,然后把此包和依賴的fastjson一起上傳到lib目錄6.4.4 編寫(xiě)方6.4.5 啟動(dòng)rootmaster flumeconf# vi mytest.conf

39、= s1a1.channels=c1 a1.sinks = r1.s1.type=itionFile = /home/flume/taildir_.s1.filegroups = f1.s1.fileHeader = true#使用自定erceptors=erceptors.i1.type=a1.channels.c1.type=a1.channels.c1.dataDirs = /home/flume/filechannle/dataDirs Dir = /home/flume/filechannle/checkpoa1.channels.c1.capacity = 1000ionCapac

40、ity=a1.sinks.r1.type=a1.sinks.r1.hdfs.path = hdfs:/master:8020/flume/spooldir a1.sinks.r1.hdfs.filePrefix =a1.sinks.r1.hdfs.round = true a1.sinks.r1.hdfs.roundValue = 10 a1.sinks.r1.hdfs.roundUnit = minute a1.sinks.r1.hdfs.fileSuffix= .log .s1.channels=c1 a1.sinks.r1.channel = c1erceptor build() ret

41、urn new Mypublicvoidconfigure(Contextcontext)rootmaster flumeconf# flume-6.4.6 測(cè)試rootmaster flumeconf# flume-6.4.6 測(cè)試echo$logrootmaster#bash7.1 說(shuō)selector：選擇作用于source階段，決定一條數(shù)據(jù)去往哪一個(gè)channel、案例演示：replicating 配置方rootmaster flumeconf# vi =a1.channels=c1=s1.r1.host=.r1.port=a1.channels.c1.keep- .r1.channel

42、s=c1 7.2.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-ng-c./conf-f-na17.2.3 測(cè)試rootmaster#echoooqiangeng|ncmaster案例演示：Multiplexing配置方rootmaster flumeconf# vi =a1.channels=c1=s1rootmaster flumeconf# vi =a1.channels=c1=s1.r1.bind=.r1.port=.USER=.ORDER=.r1.selector.default=a1.channels.c1.keep- .r1.channels=c1

43、7.3.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-ngagent-c-f./mul.conf-na17.3.3 測(cè)試.r1.channels=c1 7.3.2 啟動(dòng)agent的服務(wù)rootmasterflumeconf#flume-ngagent-c-f./mul.conf-na17.3.3 測(cè)試multiplexto #curl-X #curl-T-T-my7.4 ume的集配置配置啟動(dòng)flume-ngagent-c配置啟動(dòng)flume-ngagent-c./conf-f-再啟動(dòng)flume-ngagent-c./conf-f./conf/slave1.conf

44、 flume-ngagent-c./conf-f-測(cè)試echoiamslave1|nc21curl-T -d am7.5 ume的sinkgroups允許存放多個(gè)sinkSink sors它能夠進(jìn)行負(fù)載均衡和失敗轉(zhuǎn)移a1.sinkgroups=a1.sinkgroups.g1.sinks=k1sor.type=默認(rèn)情況下是一個(gè)sinka1.sinkgroups = g1 =k1sor.type=failover sor.priority.k1a1.sinkgroups = g1 =k1sor.type=failover sor.priority.k1=sor.priority.k2=sor.m

45、axpenalty=a1.sinkgroups = g1 =k1sor.type=load_balance sor.backoff = true sor.selector = random數(shù)據(jù)量要大，才能看出隨機(jī)的效果實(shí)現(xiàn)需求hadoop01的agent上有1個(gè)sinkgroup，里面有s1和s2兩個(gè)sink配置:hadoop01上的配置3個(gè)agent幾乎一hadoop01(a1)vi= r1 a1.channels=c1 a1.sinks=s1s2 .r1.type = http .r1.bind=hadoop01 .r1.port = 6666a1.channels.c1.type = m

46、emory ionCapacity = 10000 ercentage=20 a1.channels.c1.byteCapacity = 800000a1.sinks.s1.typeavroa1.sinks.s1.hostname=a1.sinks.s1.port=a1.sinks.s2.type = avro a1.sinks.s2.port=#sinkgroup配a1.sinkgroups = g1 =s1sor.type = failover sor.priority.s1=10sor.priority.s2=#連接三個(gè)插=a1.sinks.s1.channel=a1.sinks.s2.channel=#定義三個(gè)插=a1.sinkgroups = g1 =s1sor.type = failover sor.priority.s1=10sor.priority.s2=#連接三個(gè)插=a1.sinks.s1