大數(shù)據(jù)采集與預(yù)處理（微課版） 課件 項目5 動態(tài)網(wǎng)頁訪問日志數(shù)據(jù)采集

項目5

動態(tài)網(wǎng)頁訪問日志數(shù)據(jù)采集使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)目錄Content1使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)

2項目導(dǎo)言零

日志中通常包含了大量的用戶訪問網(wǎng)站的行為信息，比如頁面響應(yīng)時間、消耗的流量、訪問時間、停留時間、是否訪問成功等等一系列的信息，網(wǎng)站運營者可通過分析這些信息，對網(wǎng)站運營做出一定的決策。學(xué)習(xí)目標(biāo)零知識目標(biāo)了解什么是日志采集；熟悉日志數(shù)據(jù)采集的方法；掌握Flume進(jìn)行數(shù)據(jù)采集的方法；精通使用Kafka進(jìn)行消息訂閱發(fā)布的方法技能目標(biāo)具備使用Flume攔截器過濾數(shù)據(jù)的能力；具備熟悉Kafka的腳本操作的能力；具備掌握Flume采集日志文件的能力；具備掌握使用Kafka進(jìn)行消息發(fā)布訂閱的能力；素養(yǎng)目標(biāo)具備精通Flume+Kafka架構(gòu)實現(xiàn)數(shù)據(jù)采集的能力。技能目標(biāo)具備事業(yè)心和責(zé)任心；具有艱苦奮斗的精神和務(wù)實作風(fēng)；具有健康的體魄，良好的身體素質(zhì)；具備責(zé)任心。任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹Flume簡介任務(wù)技能Flume是一個由Cloudera開發(fā)的分布式、可靠和高可用的海量日志收集和傳輸系統(tǒng)，具有基于流數(shù)據(jù)的簡單靈活的架構(gòu)。Flume數(shù)據(jù)采集配置Flume實現(xiàn)數(shù)據(jù)采集的方式非常簡單，只需編輯相應(yīng)的配置文件即可完成特定數(shù)據(jù)的采集與存儲。Flume啟動啟動Flume代理使用名為flume-ng的shell腳本，該腳本位于Flume發(fā)行版的bin目錄中。Flume攔截器配置Flume能夠在數(shù)據(jù)采集的過程中對事件進(jìn)行刪除或修改，可理解為對事件進(jìn)行過濾，這些功能是靠攔截器完成的。任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹1Flume簡介Flume是一個由Cloudera開發(fā)的分布式、可靠和高可用的海量日志收集和傳輸系統(tǒng)，具有基于流數(shù)據(jù)的簡單靈活的架構(gòu)。它具有可調(diào)整的可靠性機(jī)制以及故障轉(zhuǎn)移和恢復(fù)機(jī)制，具有健壯性和容錯性。Flume-og，隨著功能的逐漸增多和完善，其存在的缺點也逐漸的暴露出來，F(xiàn)lume-og部分缺點如下：2核心組件設(shè)計不合理代碼過于臃腫14“日志傳輸”十分不穩(wěn)定核心配置缺乏標(biāo)準(zhǔn)3任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹（1）Flume三層架構(gòu)Flume由三層架構(gòu)組成，分別為agent、collector和storage。Agent層01包含F(xiàn)lume的Agent組件，與需要傳輸數(shù)據(jù)的數(shù)據(jù)源進(jìn)行連接Collector02通過多個收集器收集Agent層的數(shù)據(jù)，然后將這些轉(zhuǎn)發(fā)到下一層storage03接收collector層的數(shù)據(jù)并存儲任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹（1）Flume三層架構(gòu)Fume中包含三個核心組件分別為Source、Channel和Sink，組件說明如下所示。Sink下沉組件，負(fù)責(zé)取出Channel中的消息數(shù)據(jù)，將channel中的event數(shù)據(jù)發(fā)送到文件存儲系統(tǒng)或服務(wù)器等。采集組件，用于與數(shù)據(jù)源進(jìn)行對接，獲取數(shù)據(jù)。Source支持的數(shù)據(jù)源包括Avro、Thrift、exec、JMS、spoolingdirectory、netcat、sequencegenerator、syslog、http、legacy、自定義類型。SourceChannel是一個緩存區(qū)，用于鏈接Source和Sink組件，緩存Source寫入的Event任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹（2）Flume扇入與扇出

扇入流（數(shù)據(jù)流合并）0102Flume日志收集中常見的場景是將客戶端大量生成的日志數(shù)據(jù)發(fā)送到存儲子系統(tǒng)或一消費者代理。任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹（2）Flume扇入與扇出

扇出流（數(shù)據(jù)流復(fù)用）Flume不僅能夠同時采集多個服務(wù)器的數(shù)據(jù)到同一個服務(wù)器中，還能夠?qū)?shù)據(jù)采集并分別保存到不同服務(wù)器中。任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置

Flume實現(xiàn)數(shù)據(jù)采集的方式非常簡單，只需編輯相應(yīng)的配置文件即可完成特定數(shù)據(jù)的采集與存儲，配置文件中需要對數(shù)據(jù)源（source）、通道（channel）和接收器（sink）進(jìn)行配置。1.數(shù)據(jù)源配置

用于監(jiān)聽Avro端口并接收來自外部Avro客戶端的事件流。當(dāng)兩個Flumeagent的AvroSink和AvroSource對應(yīng)時，可組成分層集合拓?fù)浣Y(jié)構(gòu)。（1）AvroSource屬性描述channels指定通道的名稱type組件類型名稱，值為avrobind監(jiān)聽的主機(jī)名或IP地址port監(jiān)聽的端口任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置1.數(shù)據(jù)源配置

ExecSource的配置是指設(shè)置一個Linux或Unix命令，通過該命令不斷輸出數(shù)據(jù)，當(dāng)命令進(jìn)程退出時ExecSource也會退出，后續(xù)不會產(chǎn)生下一步數(shù)據(jù)。ExecSource和其他異步源的問題在于，如果將事件放入Channel失敗，無法保證數(shù)據(jù)的完整性。（2）Exec

Source屬性描述channels指定通道的名稱type組件類型名稱，值為execcommand要執(zhí)行的命令#配置一個agent，agent的名稱可自定義#指定agent的sources（如r1）、channels（如c1）a1.sources=r1a1.channels=c1a1.sources.r1.type=execmand=tail-F/var/log/securea1.sources.r1.channels=c1任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置

SpoolingDirectorySource能夠通過磁盤中的目錄攝取數(shù)據(jù)，當(dāng)監(jiān)控的制定目錄中出現(xiàn)新文件時，會將文件作為source進(jìn)行處理，一旦文件被放到“自動收集”目錄中后，便不能修改和重命名。（3）SpoolingDirectorySource屬性描述channels指定通道的名稱type組件類型名稱，值為spooldirspoolDir讀取文件的目錄。fileHeader是否添加存儲絕對路徑文件名的文件頭。#配置一個agent，agent的名稱可自定義#指定agent的sources（如r1）、channels（如c1）a1.sources=r1a1.channels=c1a1.sources.r1.type=spooldira1.sources.r1.channels=c1a1.sources.r1.spoolDir=/var/log/apache/flumeSpoola1.sources.r1.fileHeader=true任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置

Flume能夠作為Kafka的消費者，從Kafka中讀取消息，當(dāng)有多個kafka源在運行時，可以使用相同的ConsumerGroup

以便讀取一組唯一的主題分區(qū)。（4）KafkaSource屬性描述channels指定通道的名稱type組件類型名稱，值為org.apache.flume.source.kafka.KafkaSourcekafka.bootstrap.serversKafka集群中的broker列表kafka.topicsKafka消費者將從中讀取消息的以逗號分隔的主題列表。kafka.topics.regex定義源訂閱的主題集的正則表達(dá)式。此屬性的優(yōu)先級高于kafka.topicsbatchSize一批寫入Channel的最大消息數(shù)batchDurationMillis將批次寫入通道之前的最長時間（以毫秒為單位）只要達(dá)到大小和時間中的第一個，就會寫入批次。kafka.consumer.group.id消費者組的唯一標(biāo)識。任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置

監(jiān)聽指定端口并將每一行文本轉(zhuǎn)換為一個事件。與命令nc

-k

-l

[host]

[port]類似。接收的數(shù)據(jù)是以換行符分隔的文本。每一行文本都會變成一個Flume事件并通過連接的通道發(fā)送。（5）NetCatTCPSource屬性描述channels指定通道的名稱type組件類型名稱，值為spooldirbind監(jiān)聽的主機(jī)IP地址port監(jiān)聽的主機(jī)端口號#配置一個agent，agent的名稱可自定義#指定agent的sources（如r1）、channels（如c1）a1.sources=r1a1.channels=c1a1.sources.r1.type=netcata1.sources.r1.bind=a1.sources.r1.port=6666a1.sources.r1.channels=c1任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（1）MemoryChannel

使用內(nèi)存作為源與接收器之間的保留區(qū)稱之為內(nèi)存通道，內(nèi)存速度比磁盤快數(shù)倍，所以數(shù)據(jù)傳輸?shù)乃俣纫矔S之加快。2.通道配置屬性描述type組件類型名稱，需要設(shè)置為memorycapacitychannel里存放的最大event數(shù)默認(rèn)100transactionCapacityChannel每次提交的Event數(shù)量默認(rèn)100byteCapacityBufferPercentage定義byteCapacity和通道中所有事件的估計總大小之間的緩沖區(qū)百分比byteCapacity設(shè)置通道中所有事件在內(nèi)存中的大小總和任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（2）FileChannel

使用磁盤作為源與接收器之間的保留區(qū)稱之為文件通道，由于文件通道使用的是磁盤，所以會比內(nèi)存通道速度慢。文件通道優(yōu)點在于當(dāng)遇到斷電、硬件故障等能夠?qū)е翭lume采集中斷的問題時，重啟Flume即可不會造成數(shù)據(jù)丟失2.通道配置屬性描述type組件類型名稱，需要設(shè)置為filecheckpointDir存儲檢查點文件的目錄dataDirs用于存儲日志文件。在不同磁盤上使用多個目錄可以提高文件通道性能任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（1）HDFSSinks

此接收器數(shù)據(jù)寫入Hadoop分布式文件系統(tǒng)(HDFS)，持創(chuàng)建文本和序列文件，使用此接收器需要安裝Hadoop，以便Flume與HDFS集群進(jìn)行通信。3.接收器配置屬性默認(rèn)值描述channel-指定通道的名稱type-組件類型名稱，需要為hdfshdfs.path-寫入hdfs的路徑（必選）hdfs.filePrefixFLumeData文件名前綴hdfs.fileSuffix-文件名后綴hdfs.inUsePrefix-用于flume主動寫入的臨時文件的前綴hdfs.inUseSuffix.tmp用于flume主動寫入的臨時文件的后綴hdfs.rollInterval30按時間生成HDFS文件，單位：秒hdfs.rollSize1024觸發(fā)滾動的文件大小，以字節(jié)為單位(0:從不根據(jù)文件大小進(jìn)行滾動)hdfs.rollCount10按事件數(shù)量生成新文件hdfs.idleTimeout0關(guān)閉非活動文件的超時(0=禁用自動關(guān)閉空閑文件)hdfs.fileTypeSequenceFile文件格式，包括：SequenceFile,DataStream,CompressedStreamhdfs.writeFormatWritable寫文件的格式。包含：Text,Writablehdfs.roundfalse是否對時間戳四舍五入hdfs.roundValue1將其四舍五入到最高倍數(shù)hdfs.roundUnitsecond向下取整的單位，可選值包括second,minuteorhourhdfs.minBlockReplicas-設(shè)置HDFS塊的副本數(shù)，默認(rèn)采用Hadoop的配置hdfs.useLocalTimeStampFalse在替換轉(zhuǎn)義序列時使用本地時間（而不是事件標(biāo)頭中的時間戳）任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（1）HDFSSinks

HDFSSink能夠根據(jù)時間、數(shù)據(jù)大小、事件數(shù)定期滾動文件（關(guān)閉當(dāng)前文件并創(chuàng)建新文件）或按時間戳、事件起源的機(jī)器等屬性對數(shù)據(jù)進(jìn)行存儲/分區(qū)。此時在hdfs.path設(shè)置中會包含格式轉(zhuǎn)義序列，這些轉(zhuǎn)義序列將被HDFS接收器替換以生成目錄/文件名來存儲事件。3.接收器配置轉(zhuǎn)義序列描述%{host}主機(jī)名%t以毫秒為單位的Unix時間%a語言環(huán)境的簡短工作日名稱（周一、周二、...）%A語言環(huán)境的完整工作日名稱（星期一、星期二、...）%b語言環(huán)境的短月份名稱（Jan、Feb、...）%B語言環(huán)境的長月份名稱（一月，二月，...）%c語言環(huán)境的日期和時間（2005年3月3日星期四23:05:25）%d一個月中的某天(01)%e沒有填充的月份中的某一天(1)%D日期;與%m/%d/%y相同%H小時，24小時制（00…23）%I小時，12小時制%j一年中的一天(001..366)%k小時（0..23）%m月(01..12)%n簡短月份(1..12)%M分鐘(00..59)%p語言環(huán)境相當(dāng)于am或pm%s自1970-01-0100:00:00UTC以來的秒數(shù)%S秒(00..59)%y年份的最后兩位數(shù)字(00..99)%Y年（2022）轉(zhuǎn)義序列描述%{host}主機(jī)名%t以毫秒為單位的Unix時間%a語言環(huán)境的簡短工作日名稱（周一、周二、...）%A語言環(huán)境的完整工作日名稱（星期一、星期二、...）%b語言環(huán)境的短月份名稱（Jan、Feb、...）%B語言環(huán)境的長月份名稱（一月，二月，...）%c語言環(huán)境的日期和時間（2005年3月3日星期四23:05:25）%d一個月中的某天(01)%e沒有填充的月份中的某一天(1)%D日期;與%m/%d/%y相同%H小時，24小時制（00…23）%I小時，12小時制%j一年中的一天(001..366)%k小時（0..23）%m月(01..12)%n簡短月份(1..12)%M分鐘(00..59)%p語言環(huán)境相當(dāng)于am或pm%s自1970-01-0100:00:00UTC以來的秒數(shù)%S秒(00..59)%y年份的最后兩位數(shù)字(00..99)%Y年（2022）任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（2）HiveSink

HiveSink接收器能夠?qū)囟ǚ指舴奈募騄SON數(shù)據(jù)的事件流傳輸?shù)紿ive表中。由于事件是基于Hive事務(wù)編寫的，所以一旦Flume將一組事件提交到Hive后，在Hive表中可立即查詢，需要注意的是使用HiveSink時需要將Hive表創(chuàng)建為事務(wù)表，并開始Hive的事務(wù)。3.接收器配置屬性默認(rèn)值描述channel-指定通道的名稱type-組件類型名稱，需要為hivehive.metastore-HiveMetastoreURI（例如thrift://abcom:9083）hive.database-Hive數(shù)據(jù)庫名稱hive.table-Hive中的表名稱hive.partition-使用逗號分隔的分區(qū)值列表，標(biāo)識要寫入的分區(qū)。例如：如果表按(continent:string,country:string,time:string)分區(qū)，則'Asia,India,2014-02-26-01-21'將表示continent=Asia,country=India,time=2014-02-26-01-21heartBeatInterval240設(shè)置以秒為單位的心跳檢測時間間隔，設(shè)置為0表示禁用心跳檢測autoCreatePartitionstrueFlume將自動創(chuàng)建必要的Hive分區(qū)batchSize15000在單個Hive事務(wù)中寫入Hive的最大事件數(shù)maxOpenConnections500打開鏈接的最大數(shù)量，如果超過此數(shù)量，則關(guān)閉最近最少使用的連接callTimeout10000Hive和HDFSI/O操作的超時時間，單位為秒serializer

序列化器，負(fù)責(zé)從事件中解析出字段并映射到配表中的列roundUnitminute向下取整的單位，可選值包括second,minuteorhourroundValue1向下舍入到此的最高倍數(shù)（在使用hive.roundUnit配置的單位中），小于當(dāng)前時間timeZoneLocalTime用于解析分區(qū)中的轉(zhuǎn)義序列的時區(qū)名稱，例如America/Los_AngelesuseLocalTimeStampFalse在替換轉(zhuǎn)義序列時使用本地時間（而不是事件標(biāo)頭中的時間戳）任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（3）AvroSink

該接收器與AvroSource可構(gòu)成分層架構(gòu)。發(fā)送到此接收器的Flume事件將轉(zhuǎn)換為Avro事件并發(fā)送到配置的主機(jī)名/端口。3.接收器配置屬性默認(rèn)值描述channel-指定通道的名稱type-組件類型名稱，需要為avrohostname-監(jiān)聽的主機(jī)名或IP地址port-監(jiān)聽的端口號。任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹2Flume數(shù)據(jù)采集配置（4）FileRollSink

該接收器能夠?qū)⑹录鎯υ诒镜匚募到y(tǒng)上。3.接收器配置屬性默認(rèn)值描述channel-指定通道的名稱type-組件類型名稱，需要為file_rollsink.directory-將存儲文件的目錄sink.rollInterval30每30秒滾動一次文件。指定0將禁用滾動并導(dǎo)致將所有事件寫入單個文件任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹3Flume啟動啟動Flume代理使用名為flume-ng的shell腳本，該腳本位于Flume發(fā)行版的bin目錄中。啟動時需要在命令中指定代理名稱、配置目錄以及配置文件，并且設(shè)置的代理名稱要和配置文件中的代理名稱一致，啟動數(shù)據(jù)采集的命令如下。bin/flume-ngagent--nameagent_name--confconf–conf-fileconf/perties.template任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹3Flume啟動參數(shù)描述--name指定Agent的名稱（必選）--conf指定配置文件所在目錄--conf-file指定配置文件任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹3Flume啟動啟動Flumeagent進(jìn)行數(shù)據(jù)采集，命令如下所示。[root@masterflume-code]#cd/usr/local/flume/bin[root@masterbin]#./flume-ngagent--confconf--conf-file/usr/local/inspur/code/flume-code/example.conf--namea1任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹3Flume啟動

單獨啟動另一個終端，使用telnet向4444向Flume發(fā)送事件，啟動成功后輸入“Helloworld！”，命令如下所示。[root@master~]#telnetlocalhost4444任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹4Flume攔截器配置

時間攔截器（TimestampInterceptor）1靜態(tài)攔截器（StaticInterceptor）3正則表達(dá)式過濾攔截器（RegexFilteringInterceptor）5主機(jī)攔截器（HostInterceptor）2搜索和替換攔截器（SearchandReplaceInterceptor）4任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹4Flume攔截器配置（1）時間攔截器（TimestampInterceptor）時間攔截器將處理事件的時間（以毫秒為單位）插入到事件頭中。并根據(jù)時間戳將數(shù)據(jù)寫入到不同文件中，當(dāng)不使用任何攔截器時，F(xiàn)lume接收到的只有message。屬性默認(rèn)值描述type-組件類型名稱，值為timestampheaderNametimestamp生成的時間戳的標(biāo)頭的名稱preserveExistingfalse如果時間戳已經(jīng)存在，是否應(yīng)該保留值為true或falsea1.sources=r1a1.channels=c1a1.sources.r1.channels=c1a1.sources.r1.type=avroerceptors=i1erceptors.i1.type=timestamp任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹4Flume攔截器配置（2）主機(jī)攔截器（HostInterceptor）

該攔截器向事件中添加包含當(dāng)前Flume代理的主機(jī)名或IP地址，主要表現(xiàn)形式為在HDFS中顯示以Flume主機(jī)IP地址做為前綴的文件名。屬性默認(rèn)值描述type-組件類型名稱，值為hostpreserveExistingfalse如果主機(jī)頭已經(jīng)存在是否保留，值為true或falseuseIPtrue如果為true，請使用IP地址，否則使用主機(jī)名hostHeaderhost要使用的文件頭a1.sources=r1a1.channels=c1erceptors=i1erceptors.i1.type=host任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹4Flume攔截器配置（3）靜態(tài)攔截器（StaticInterceptor）靜態(tài)攔截器能夠?qū)⒕哂徐o態(tài)值的靜態(tài)事件頭附加到所有事件，同一個靜態(tài)攔截器中不能夠設(shè)置多個事件頭，但可以設(shè)置多個靜態(tài)攔截器。屬性默認(rèn)值描述type-組件類型名稱，值為staticpreserveExistingtrue配置的標(biāo)頭已經(jīng)存在，是否應(yīng)該保留。值為true或falsekeykey事件頭名稱valuevalue靜態(tài)值a1.sources=r1a1.channels=c1a1.sources.r1.channels=c1a1.sources.r1.type=avroerceptors=i1erceptors.i1.type=staticerceptors.i1.key=datacentererceptors.i1.value=NEW_YORK任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹4Flume攔截器配置（4）搜索和替換攔截器（SearchandReplaceInterceptor）該攔截器提供了基于Java正則表達(dá)式的字符串的搜索和替換功能。屬性默認(rèn)值描述type-組件類型名稱，值為search_replacesearchPatterntrue要搜索和替換的模式replaceStringkey替換字符串charsetvalue事件主體的字符集。默認(rèn)情況下假定為UTF-8erceptors=search-replaceerceptors.search-replace.type=search_replace#刪除事件正文中的前導(dǎo)字母數(shù)字字符。erceptors.search-replace.searchPattern=^[A-Za-z0-9_]+erceptors.search-replace.replaceString=任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹4Flume攔截器配置（5）正則表達(dá)式過濾攔截器（RegexFilteringInterceptor）該攔截器通過將事件主體解釋為文本，并將文本與配置的正則表達(dá)式匹配完成事件的過濾。屬性默認(rèn)值描述type-組件類型名稱，值為regex_filterregex“*”用于匹配事件的正則表達(dá)式excludeEventsfalse如果為true，則正則表達(dá)式確定要排除的事件，否則正則表達(dá)式確定要包含的事件。erceptors=i1erceptors.i1.type=regex_filtererceptors.i1.regex=(\\d):(\\d):(\\d)安裝啟動httpdf服務(wù)Step1啟動FlumeStep2瀏覽器訪問httpd服務(wù)器Step3查看采集到本地文件系統(tǒng)的數(shù)據(jù)Step4訪問日志數(shù)據(jù)使用Flume的扇出架構(gòu)，采集httpd服務(wù)器日志數(shù)據(jù)到HDFS分布式文件系統(tǒng)和本地文件系統(tǒng)任務(wù)5-1使用Flume采集某官網(wǎng)訪問日志數(shù)據(jù)壹任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳1Kafka簡介Kafka是由Apache軟件基金會開發(fā)的一個開源流處理平臺。是一個快速、可擴(kuò)展的、分布式的、分區(qū)的、可復(fù)制的基于zookeeper協(xié)調(diào)的分布式日志系統(tǒng)，用于web/nginx日志、訪問日志、消息服務(wù)等，Linkedin于2010年將Kafka貢獻(xiàn)給了Apache基金會并成為頂級開源項目。持久性、可靠性高并發(fā)高吞吐量、低延遲容錯性可擴(kuò)展性23451任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳1Kafka簡介任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳2Kafka配置1將zookeeper安裝包上傳到“/usr/local”目錄下并解壓重命名為zookeeper2將模板文件“zoo_sample.cfg”文件復(fù)制重命名為“zoo.cfg”3進(jìn)入zookeeper安裝目錄的“bin”目錄，啟動zookeeper并查看進(jìn)程5在Kafka的“/bin”目錄下后臺啟動Kafka進(jìn)程4將Kafka安裝包上傳到“/usr/local”目錄下解壓并重命名為kafka任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳3Kafka腳本操作Kafka為開發(fā)人員提供了眾多腳本對Kafka進(jìn)行管理和相關(guān)的操作，其中常用的腳本包含Topic腳本、生產(chǎn)者腳本、消費者腳本等。任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳3Kafka腳本操作1.Topic腳本Topic腳本文件名為“kafka-topic.sh”，通過該腳本文件能夠完成創(chuàng)建主題、查看主題、修改主題、刪除主題等操作。參數(shù)描述--alter修改主題--config<String:name=value>創(chuàng)建或修改主題時，用于設(shè)置主題級別的參數(shù)--create創(chuàng)建主題--delete刪除主題--describe查看主題的詳細(xì)信息--disable-rack-aware創(chuàng)建主題時不考慮機(jī)架信息--help打印幫助信息--if-exists修改或刪除主題時使用，只有當(dāng)主題存在時才會執(zhí)行動作--if-not-exists創(chuàng)建主題時使用，只有主題不存在時才會執(zhí)行動作--list列出所有可用的主題--partitons分區(qū)數(shù)創(chuàng)建主題或增加分區(qū)時指定的分區(qū)數(shù)--replica-assignment分配方案手工指定分區(qū)副本分配方案--replication-factor副本數(shù)創(chuàng)建主題時指定副本數(shù)--topic主題名稱指定主題名稱--topics-with-overrides使用describe查看主題信息時，只展示包含覆蓋配置的主題--unavailable-partitions使用describe查看主題信息時，只展示包含沒有l(wèi)eader副本的分區(qū)--under-replicated-partitions使用describe查看主題信息時，只展示包含失效副本的分區(qū)--bootstrap-server鏈接的服務(wù)器地址和端口（必需）任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳3Kafka腳本操作2.生產(chǎn)者腳本

生產(chǎn)者腳本文件名為“kafka-console-producer.sh”，該腳本能夠生產(chǎn)創(chuàng)建消息并將消息發(fā)送到消費者。參數(shù)描述--bootstrap-server要連接的服務(wù)器必需(除非指定–broker-list)--topic(必需)topic名稱--batch-size單個批處理中發(fā)送的消息數(shù)（默認(rèn)值：16384）--compression-codec壓縮編解碼器，可選值“none”、“gzip”、“snappy”、“l(fā)z4”或“zstd”。如果未指定值，則默認(rèn)為“gzip”--max-block-ms在發(fā)送請求期間，生產(chǎn)者將阻止的最長時間（默認(rèn)值：60000）--max-memory-bytes生產(chǎn)者用來緩沖等待發(fā)送到服務(wù)器的總內(nèi)存(默認(rèn)值:33554432)--max-partition-memory-bytes為分區(qū)分配的緩沖區(qū)大小(默認(rèn)值:16384)--message-send-max-retries最大的重試發(fā)送次數(shù)(默認(rèn)值:3)--metadata-expiry-ms強制更新元數(shù)據(jù)的時間閾值(ms)(默認(rèn)值:300000)--producer-property將自定義屬性傳遞給生成器的機(jī)制--producer.config生產(chǎn)者配置屬性文件[--producer-property]優(yōu)先于此配置配置文件完整路徑--property自定義消息讀取器--request-required-acks生產(chǎn)者請求的確認(rèn)方式--request-timeout-ms生產(chǎn)者請求的確認(rèn)超時時間--retry-backoff-ms生產(chǎn)者重試前，刷新元數(shù)據(jù)的等待時間閾值--socket-buffer-sizeTCP接收緩沖大小--timeout消息排隊異步等待處理的時間閾值--sync同步發(fā)送消息任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳3Kafka腳本操作3.消費者腳本

消費者腳本文件名為“kafka-console-consumer.sh”，該腳本接收并消費生產(chǎn)者發(fā)送的消息。參數(shù)描述--group指定消費者所屬組的ID--topic被消費的topic--partition指定分區(qū)；除非指定–offset，否則從分區(qū)結(jié)束(latest)開始消費--offset執(zhí)行消費的起始o(jì)ffset位置;默認(rèn)值:latest;/latest/earliest/偏移量--whitelist正則表達(dá)式匹配topic；–topic就不用指定了;匹配到的所有topic都會消費;當(dāng)然用了這個參數(shù),--partition--offset等就不能使用了--consumer-property將用戶定義的屬性以key=value的形式傳遞給使用者--consumer.config消費者配置屬性文件請注意，[consumer-property]優(yōu)先于此配置--property初始化消息格式化程序的屬性--from-beginning從存在的最早消息開始，而不是從最新消息開始,注意如果配置了客戶端名稱并且之前消費過，那就不會從頭消費了--max-messages消費的最大數(shù)據(jù)量，若不指定，則持續(xù)消費下去--skip-message-on-error如果處理消息時出錯，請?zhí)^它而不是暫停--isolation-level設(shè)置為read_committed以過濾掉未提交的事務(wù)性消息,設(shè)置為read_uncommitted以讀取所有消息,默認(rèn)值:read_uncommitted任務(wù)5-2使用Kafka消費Flume中某官網(wǎng)訪問日志數(shù)據(jù)貳4KafkaPythonAPI生產(chǎn)者API顧名思義，就是指應(yīng)用程序發(fā)送數(shù)據(jù)流到Kafka集群中的topic中，供消費者消費使用，生產(chǎn)者是線程安全的，跨線程共享單個生產(chǎn)者實例通常比擁有多個實例更快。