課程概述課程性質(zhì)：專業(yè)核心課。前置相關(guān)課程：Linux系統(tǒng)管理與維護(hù)、大數(shù)據(jù)集群搭建維護(hù)與數(shù)據(jù)存儲、Java程序設(shè)計、Python編程基礎(chǔ)、關(guān)系型數(shù)據(jù)庫管理與設(shè)計、NoSQL數(shù)據(jù)庫基礎(chǔ)與應(yīng)用課程目標(biāo)：掌握大數(shù)據(jù)的采集與存儲架構(gòu)、數(shù)據(jù)轉(zhuǎn)換的知識，具備對系統(tǒng)數(shù)據(jù)的收集與存儲，數(shù)據(jù)遷移工具的使用、數(shù)據(jù)處理的能力。學(xué)時：理論32學(xué)時，實踐64學(xué)時實踐用工具：VMWareWorkStationPro15.5.1SecureCRTx648.5.4EclipseIDEforJavaDevelopers-2020-12VisualStudioCode1.57.1實踐用硬件環(huán)境：CPU：≥inteli3系列內(nèi)存：≥16G

固態(tài)硬盤（建議）：≥128G實踐用軟件環(huán)境：LinuxCentOS7.x集群JDK版本：Java64-Bit1.8Python版本：Python3.7.9Hadoop3.3.x高可用集群

三個節(jié)點：node1（主）、node2（主）、node3（從）第一單元數(shù)據(jù)采集Flume掌握Flume的安裝與配置掌握Flume采集方案的編寫熟悉Flume三大組件的作用熟悉Flume三大組件的常用類型1.1什么是Flume

Flume是一個進(jìn)行海量流式（streaming）事件數(shù)據(jù)采集的分布式系統(tǒng)。Flume多用于對軟件系統(tǒng)的日志文件數(shù)據(jù)的采集，目前成為Apache軟件基金會的頂級項目之一。Flume可以將應(yīng)用產(chǎn)生的數(shù)據(jù)存儲到任何集中存儲器中，比如HDFS,HBase數(shù)據(jù)緩存。當(dāng)收集數(shù)據(jù)的速度超過將寫入數(shù)據(jù)的時候，保證其能夠在兩者之間提供平衡。Flume的管道是基于事務(wù)，保證了數(shù)據(jù)在傳送和接收時的一致性.Flume是可靠的，容錯性高的，可升級的，易管理的,并且可定制的1.2Flume的主要組件

Flume采集系統(tǒng)以Agent為單位，一個Agent就是一個工作進(jìn)程。一個完整的Agent由三個組件構(gòu)成，它們分別是：(1)Source：用于從數(shù)據(jù)源接收數(shù)據(jù)（即采集數(shù)據(jù)），需要設(shè)置數(shù)據(jù)的來源類型；(2)Sink：用于傳遞數(shù)據(jù)給目的地（即保存數(shù)據(jù)），需要設(shè)置數(shù)據(jù)的目的地類型；(3)Channel：用于Source和Sink的連接，并緩存?zhèn)鬏數(shù)臄?shù)據(jù)，需要設(shè)置緩存類型。Source常用類型序號類型名稱功能1spooldir數(shù)據(jù)來源于磁盤文件目錄2exec表明數(shù)據(jù)來源于Linux命令3taildir表明數(shù)據(jù)來源于文件的實時新增內(nèi)容4avro表明數(shù)據(jù)來源于指定IP和端口發(fā)送的AVRO消息5thrift表明數(shù)據(jù)來源于指定IP和端口發(fā)送的THRIFT消息6netcat表明數(shù)據(jù)來源于指定IP和端口發(fā)送的netcatTCP消息7netcatudp表明數(shù)據(jù)來源于指定IP和端口發(fā)送的netcatUDP消息Channel常用類型序號類型名稱功能1memory以內(nèi)存作為數(shù)據(jù)的緩存，讀寫速度快，但數(shù)據(jù)可能會丟失2file以磁盤文件作為數(shù)據(jù)的緩存，讀寫速度相對慢，但數(shù)據(jù)不會丟失Sink常用類型序號類型名稱功能1hdfs2file_roll表明數(shù)據(jù)的目的地是系統(tǒng)的本地目錄3KafkaSink表明數(shù)據(jù)的目的地是Kafka的生產(chǎn)者4avro表明采用的是AVRO消息方式傳遞數(shù)據(jù)到數(shù)據(jù)目的地，需指定數(shù)據(jù)目的地的主機(jī)名和端口5hbase表明數(shù)據(jù)的目的地是HBase6asynchbase表明以異步方式寫數(shù)據(jù)到HBase7hive表明數(shù)據(jù)的目的地是Hive8elasticsearch表明數(shù)據(jù)的目的地是elasticsearch1.3Flume的安裝與操作

1.3.1安裝Flume

在瀏覽器的地址欄中輸入Flume的官網(wǎng)地址“/download.html”，在打開的頁面中，單擊鏈接，打開Flume的下載鏡像地址頁面。在打開后的Flume的下載鏡像地址頁面，復(fù)制Flume的鏡像下載地址。

下載Flume安裝包到此目錄，以下代碼中所列地址只作格式說明，不能作為實際下載地址使用，實際地址從官網(wǎng)中復(fù)制。[root@node1software]#wget/apache/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz將flume的“conf”目錄下的“flume-env.sh.template”文件，復(fù)制重命名。[root@node1conf]#cpflume-env.sh.templateflume-env.sh在文件中追加以下內(nèi)容。exportFLUME_HOME=/usr/app/apache-flume-1.9.0-bin/exportFLUME_CONF_DIR=$FLUME_HOME/confexportPATH=$PATH:$FLUME_HOME/bin打開復(fù)制重命名后的“flume-env.sh”文件。[root@node1conf]#viflume-env.sh向文件中添加JAVA的安裝位置，以下代碼只作格式說明，實際安裝位置根據(jù)自己安裝的JAVA實際路徑填寫。exportJAVA_HOME=/usr/java/jdk打開系統(tǒng)的環(huán)境配置文件。[root@node1conf]#vi/etc/profile1.3.2單機(jī)操作Flume

單機(jī)操作Flume指的是只使用集群中的一個節(jié)點進(jìn)行數(shù)據(jù)的采集。但不論采用單機(jī)還是集群操作Flume，F(xiàn)lume的采集方案中至少要配置有一個Agent，一個Agent中可以包含多個Source、Channel和Sink，一個Source可以指定多個Channel，但是一個Sink只能綁定一個Channel。創(chuàng)建新文件file-logger.conf，編寫采集方案。[root@node1flume-test]#vifile-logger.conf在文件中寫入。#定義Agent中各組件名稱，Agent名為a1a1.sources=r1#sources名為r1a1.sinks=k1#sinks名為k1a1.channels=c1#channels名為c1a1.sources.r1.type=spooldir#配置Sources的數(shù)據(jù)源類型a1.sources.r1.spoolDir=/usr/test/flume-test/1/input

a1.sinks.k1.type=file_roll#配置Sink的數(shù)據(jù)目的地類型a1.sinks.k1.sink.directory=/usr/test/flume-test/1/output

a1.channels.c1.type=memory#配置緩存類型為內(nèi)存模式a1.channels.c1.capacity=1000#指定緩存的最大值a1.channels.c1.transactionCapacity=100#指定緩存在每次事務(wù)中，讀source或?qū)憇ink的最大值#把Source和Sink綁定到Channel上a1.sources.r1.channels=c1a1.sinks.k1.channel=c1根據(jù)數(shù)據(jù)源類型，配置數(shù)據(jù)源的位置1.單點采集數(shù)據(jù)到本地根據(jù)數(shù)據(jù)目的地類型，配置目的地位置執(zhí)行以下命令啟動Flume，進(jìn)程一直監(jiān)控“input”目錄的變化。[root@node11]#flume-ngagent-cconf/-ffile-logger.conf-na1

-Dflume.root.logger=INFO,console將日志素材文件access.log，復(fù)制到“input”目錄。復(fù)制后，F(xiàn)lume采集進(jìn)程會監(jiān)控到“input”目錄的變化，開始執(zhí)行采集，采集后，輸出如下反饋信息。[root@node11]#INFOavro.ReliableSpoolingFileEventReader:Preparingtomovefile/usr/test/flume-test/1/input/access.logto/usr/test/flume-test/1/input/access.log.COMPLETED2.單點采集數(shù)據(jù)到HDFS創(chuàng)建新文件“hdfs-logger.conf”，編寫采集方案。[root@node1flume-test]#vihdfs-logger.conf向文件中寫入。.......#配置Sinka1.sinks.k1.type=hdfs#指定采集后的數(shù)據(jù)存放到hdfs的路徑，此時的節(jié)點node1應(yīng)處于Active狀態(tài)，#用時間格式符產(chǎn)生路徑名稱，用以區(qū)分采集后的數(shù)據(jù)。a1.sinks.k1.hdfs.path=hdfs://node1:9820/flume/2/data/%Y-%m-%d/%H%M#設(shè)置間隔多少秒滾動生成一個HDFS文件，0表示不按時間長度滾動生成文件，默認(rèn)值是30秒a1.sinks.k1.hdfs.rollInterval=0#數(shù)據(jù)達(dá)到多少字節(jié)時滾動生成一個新的HDFS文件，如果是0，則不根據(jù)大小滾動生成文件，默認(rèn)值是1024a1.sinks.k1.hdfs.rollSize=10240000a1.sinks.k1.hdfs.rollCount=0#0表示不基于事件數(shù)量值滾動生成文件，默認(rèn)值是10#設(shè)置無響應(yīng)狀態(tài)下的超時時間（秒），默認(rèn)值是0，表示不關(guān)閉無響應(yīng)狀態(tài)下的文件a1.sinks.k1.hdfs.idleTimeout=3未完指定采集后的數(shù)據(jù)目的地類型，及相關(guān)配置#輸出的文件格式，DataStream表示不壓縮，SequenceFile表示序列格式，#CompressedStream表示壓縮格式，a1.sinks.k1.hdfs.fileType=DataStream#但需要額外設(shè)置hdfs.codeC屬性，指定壓縮類型a1.sinks.k1.hdfs.round=true#時間戳是否應(yīng)該向下舍入#時間戳是否應(yīng)該向下舍入時，四舍五入到該值的最高倍數(shù)，小于當(dāng)前時間a1.sinks.k1.hdfs.roundValue=10a1.sinks.k1.hdfs.roundUnit=minute#舍入值的單位，可以是second、minute或houra1.sinks.k1.hdfs.useLocalTimeStamp=true#使用本地時間戳......接上頁在瀏覽器的地址欄中輸入HDFS的地址“node1:9870”，依次單擊“flume”→“2”→“data”→“日期”→“時間”目錄，最后在“時間”子目錄中，會顯示有存放Flume采集結(jié)果的兩個文件。3.采集持續(xù)增加的內(nèi)容，保存到HDFS......#配置Sourcesa1.sources.r1.type=exec#a1.sources.r1.spoolDir=/usr/test/flume-test/3/input#指定shell命令，以及持續(xù)監(jiān)測的目標(biāo)文件mand=tail-F/usr/test/flume-test/3/input/test.log......#配置Channel，使用文件作為緩存a1.channels.c1.type=file#設(shè)置channel檢查點文件的存放目錄a1.channels.c1.checkpointDir=/usr/test/flume-test/3/dataCheckpointDira1.channels.c1.dataDirs=/usr/test/flume-test/3/data#設(shè)置channel日志文件的存放目錄數(shù)據(jù)源類型第一種方式（使用Linux命令的方式）：......#配置Sourcesa1.sources.r1.type=TAILDIR#a1.sources.r1.spoolDir=/usr/test/flume-test/4/input#以json格式記錄每次查詢新增內(nèi)容時的最后位置，以及查詢的絕對路徑和節(jié)點a1.sources.r1.positionFile=/usr/test/flume-test/4/record/taildir_position.jsona1.sources.r1.filegroups=f1f2a1.sources.r1.filegroups.f1=/usr/test/flume-test/4/input/test.txt#指定組1的目標(biāo)文件#指定組2的目標(biāo)文件，文件名中采用了通配符a1.sources.r1.filegroups.f2=/usr/test/flume-test/4/input/.*log.*......數(shù)據(jù)源類型如果要監(jiān)控多個文件，可以設(shè)置文件組，按組的方式，管理文件第二種方式（文件的實時新增內(nèi)容方式）：1.3.3Flume集群的搭建

在多個節(jié)點中安裝Flume，組建成Flume集群。組建過程可以利用節(jié)點node1中已安裝好的Flume，直接復(fù)制到其它節(jié)點中，具體步驟如下：將主節(jié)點node1上的環(huán)境變量文件復(fù)制到其它兩個節(jié)點。[root@node1~]#scp/etc/profileroot@node2:/etc/[root@node1~]#scp/etc/profileroot@node3:/etc/將主節(jié)點node1上安裝好的Flume復(fù)制到其它兩個節(jié)點。[root@node1~]#scp-r/usr/app/apache-flume-1.9.0-bin/root@node2:/usr/app/[root@node1~]#scp-r/usr/app/apache-flume-1.9.0-bin/root@node3:/usr/app/在其它兩個節(jié)點上，執(zhí)行下列命令，使更新后的環(huán)境變量生效。source/etc/profile1.3.4Flume集群模式下的數(shù)據(jù)采集

場景：集群中三個節(jié)點分工負(fù)責(zé)，共同完成數(shù)據(jù)的采集。采集模型見下圖，在這個模型中，節(jié)點node1負(fù)責(zé)采集數(shù)據(jù)，節(jié)點node1與節(jié)點node2、node3之間采用avro協(xié)議來傳遞消息；節(jié)點node2、node3負(fù)責(zé)從節(jié)點node1接收數(shù)據(jù)，并將數(shù)據(jù)傳遞給數(shù)據(jù)目的地；由于節(jié)點node1需要對接node2、node3兩個節(jié)點，因此在節(jié)點node1上需要設(shè)置兩個sink，組成一個sinkgroup，兩個sink之間的工作協(xié)同由sinkgroup負(fù)責(zé)完成。

sinkgroup通過sinks屬性來設(shè)置sinkgroup的成員，通過processor屬性的設(shè)置，來實現(xiàn)多個sink之間的工作協(xié)調(diào)，從而達(dá)到提升Flume集群工作的可靠性效果。sinkgroupprocessor的常用類型見下表。序號類型功能配套屬性功能1Load_balance用于sinkgroup成員間工作壓力的負(fù)載均衡backoff布爾類型，true值表示將有故障的成員列入禁用黑名單。selector用于設(shè)置成員間的協(xié)調(diào)機(jī)制，其值有三種選項：round_robin表示采用輪詢的方式；random表示采用隨機(jī)的方式；AbstractSinkSelector表示自定義的方式。maxTimeOut用于設(shè)置故障成員被禁用的最長時間，毫秒級。2failover對有故障的sinkgroup成員進(jìn)行管理與維護(hù)priority用于設(shè)置成員的優(yōu)先級，數(shù)值越大，優(yōu)先級越高。優(yōu)先選擇優(yōu)先級最高的成員處理數(shù)據(jù)。maxpenalty用于設(shè)置故障成員的最大重試間隔時間，以毫秒為時間單位。1.主節(jié)點的設(shè)置#配置負(fù)載均衡模式下的一級采集方案a1.sources=r1a1.sinks=k1k2a1.channels=c1#配置sources組件a1.sources.r1.type=TAILDIR#以json格式記錄每次查詢新增內(nèi)容時的最后位置，以及查詢的絕對路徑和節(jié)點a1.sources.r1.positionFile=/usr/test/flume-test/group01/record/taildir_position.jsona1.sources.r1.filegroups=f1#為了簡化，只設(shè)置1個文件組a1.sources.r1.filegroups.f1=/usr/test/flume-test/group01/input/test.txt#指定組1的采集文件#設(shè)置channelsa1.channels.c1.type=memorya1.channels.c1.capacity=1000a1.channels.c1.transactionCapacity=100未完配置兩個sink接上一頁a1.sinks.k1.channel=c1a1.sinks.k1.type=avro#設(shè)置sink的類型為avro消息a1.sinks.k1.hostname=node2#指定接收avro消息的主機(jī)名a1.sinks.k1.port=8888#指定avro通訊的端口號a1.sinks.k2.channel=c1a1.sinks.k2.type=avroa1.sinks.k2.hostname=node3a1.sinks.k2.port=8888a1.sinkgroups=g1#設(shè)置集群內(nèi)的負(fù)載均衡方案a1.sinkgroups.g1.sinks=k1k2cessor.type=load_balance#設(shè)置processor為load_balance，實現(xiàn)成員間的負(fù)載均衡cessor.backoff=truecessor.selector=randomcessor.maxTimeOut=10000將第1個sink綁定到node2節(jié)點將第2個sink綁定到node3節(jié)點設(shè)置sinkgroup的成員2.從節(jié)點的設(shè)置#配置負(fù)載均衡模式下的二級采集方案的一個sink分支a1.sources=r1a1.sinks=k1a1.channels=c1#配置sources組件a1.sources.r1.type=avro#設(shè)置source的類型為avro消息a1.sources.r1.bind=node2a1.sources.r1.port=8888#指定avro通訊的端口號a1.sinks.k1.type=hdfs#數(shù)據(jù)的目的地類型......綁定本機(jī)主機(jī)名，如節(jié)點2，設(shè)置為node2，節(jié)點3，設(shè)置為node33.啟動Flume集群：節(jié)點node2、node3的FlumeAgent啟動應(yīng)先于主節(jié)點node1。在節(jié)點node2、node3的對應(yīng)目錄下，分別啟動FlumeAgent，然后啟動主節(jié)點node1上的FlumeAgent（三個節(jié)點上所用命令的內(nèi)容一致）。1.3.5復(fù)雜場景下的Flume集群數(shù)據(jù)采集

1.復(fù)雜場景服務(wù)器node2、node3實時產(chǎn)生日志數(shù)據(jù)，日志數(shù)據(jù)有admin.log、user.log、orders.log文件，現(xiàn)在需要將服務(wù)器node2、node3上產(chǎn)生的這三類數(shù)據(jù)采集到服務(wù)器node1上，存于HDFS。2.采集方案設(shè)計為實現(xiàn)上面的場景需求，將設(shè)計節(jié)點node2、node3負(fù)責(zé)采集產(chǎn)生的日志數(shù)據(jù)，節(jié)點node1負(fù)責(zé)匯總數(shù)據(jù)到HDFS，并借助Flume的interceptor（攔截器）對數(shù)據(jù)進(jìn)行標(biāo)記，根據(jù)標(biāo)記進(jìn)行數(shù)據(jù)的歸類。采集模型見下一頁。Flume的interceptor主要用于修改或刪除event的header信息，interceptor的常用類型及屬性見教材中的表1.5。采集模型(1)在node2和node3節(jié)點編寫采集方案#配置Agent組件，編寫第一級日志采集方案a1.sources=r1r2r3a1.sinks=k1a1.channels=c1#配置第1個sources組件a1.sources.r1.type=TAILDIR......erceptors=i1i2erceptors.i1.type=static#第1個攔截器的類型為static#設(shè)置鍵名為type，注意這個鍵名type會在下游的agent中被作為變量形式%{type}被調(diào)用erceptors.i1.key=typeerceptors.i1.value=admin#第1個攔截器的類型為timestamp，會在下游的agent中被作為變量形式%Y-%m-%d被調(diào)用erceptors.i2.type=timestamp未完配置3個Source，用于對應(yīng)采集三類日志數(shù)據(jù)文件為第1個source配置兩個攔截器，名稱分別為i1和i2，名稱之間用英文的空格隔開接上一頁#配置第2個sources組件a1.sources.r2.type=TAILDIR......#為第2個source配置兩個攔截器，名稱分別為i1和i2，名稱之間用英文的空格間隔erceptors=i1i2erceptors.i1.type=staticerceptors.i1.key=typeerceptors.i1.value=usererceptors.i2.type=timestamp#配置第3個sources組件a1.sources.r3.type=TAILDIR......#為第3個source配置