兩臺 ima jar 寫入 組裝 ESS 操作 采集 key-value

1． 日誌的采集和匯總
1.1． 案例場景
A、B兩臺日誌服務機器實時生產日誌主要類型為access.log、nginx.log、web.log
現在要求：
把A、B 機器中的access.log、nginx.log、web.log 采集匯總到C機器上然後統一收集到hdfs中。
但是在hdfs中要求的目錄為：
/source/logs/access/20160101/
/source/logs/nginx/20160101/
/source/logs/web/20160101/**
1.2． 場景分析

1.3． 數據流程處理分析

1.4． 功能實現
①　在服務器A和服務器B上
創建配置文件 exec_source_avro_sink.conf

Name the components on this agent

a1.sources = r1 r2 r3
a1.sinks = k1
a1.channels = c1

Describe/configure the source

a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/data/access.log
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static

static攔截器的功能就是往采集到的數據的header中插入自## 己定義的key-value對

a1.sources.r1.interceptors.i1.key = type

a1.sources.r1.interceptors.i1.value = access
a1.sources.r2.type = exec
a1.sources.r2.command = tail -F /root/data/nginx.log
a1.sources.r2.interceptors = i2
a1.sources.r2.interceptors.i2.type = static
a1.sources.r2.interceptors.i2.key = type
a1.sources.r2.interceptors.i2.value = nginx
a1.sources.r3.type = exec
a1.sources.r3.command = tail -F /root/data/web.log
a1.sources.r3.interceptors = i3
a1.sources.r3.interceptors.i3.type = static
a1.sources.r3.interceptors.i3.key = type
a1.sources.r3.interceptors.i3.value = web

Describe the sink

a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.200.101
a1.sinks.k1.port = 41414

Use a channel which buffers events in memory

a1.channels.c1.type = memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity = 10000

Bind the source and sink to the channel

a1.sources.r1.channels = c1
a1.sources.r2.channels = c1
a1.sources.r3.channels = c1
a1.sinks.k1.channel = c1
②　在服務器C上創建配置文件 avro_source_hdfs_sink.conf 文件內容為
#定義agent名， source、channel、sink的名稱
a1.sources = r1
a1.sinks = k1
a1.channels = c1
#定義source
a1.sources.r1.type = avro
a1.sources.r1.bind = mini2
a1.sources.r1.port =41414
#添加時間攔截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type =
org.apache.flume.interceptor.TimestampInterceptor$Builder
#定義channels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity = 10000
#定義sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path=hdfs://192.168.200.101:9000/source/logs/%{type}/%Y%m%d
a1.sinks.k1.hdfs.filePrefix =events
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
#時間類型
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件不按條數生成
a1.sinks.k1.hdfs.rollCount = 0
#生成的文件按時間生成
a1.sinks.k1.hdfs.rollInterval = 30
#生成的文件按大小生成
a1.sinks.k1.hdfs.rollSize = 10485760
#批量寫入hdfs的個數
a1.sinks.k1.hdfs.batchSize = 10000
flume操作hdfs的線程數（包括新建，寫入等）
a1.sinks.k1.hdfs.threadsPoolSize=10
#操作hdfs超時時間
a1.sinks.k1.hdfs.callTimeout=30000
#組裝source、channel、sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
③　配置完成之後，在服務器A和B上的/root/data有數據文件access.log、nginx.log、web.log。先啟動服務器C上的flume，啟動命令
在flume安裝目錄下執行 ：
bin/flume-ng agent -c conf -f conf/avro_source_hdfs_sink.conf -name a1 -Dflume.root.logger=DEBUG,console
然後在啟動服務器上的A和B，啟動命令
在flume安裝目錄下執行 ：
bin/flume-ng agent -c conf -f conf/exec_source_avro_sink.conf -name a1 -Dflume.root.logger=DEBUG,console

2． Flume自定義攔截器（了解）
2.1． 案例背景介紹
Flume是Cloudera提供的一個高可用的，高可靠的，分布式的海量日誌采集、聚合和傳輸的系統，Flume支持在日誌系統中定制各類數據發送方，用於收集數據；同時，Flume提供對數據進行簡單處理，並寫到各種數據接受方（可定制）的能力。Flume有各種自帶的攔截器，比如：TimestampInterceptor、HostInterceptor、RegexExtractorInterceptor等，通過使用不同的攔截器，實現不同的功能。但是以上的這些攔截器，不能改變原有日誌數據的內容或者對日誌信息添加一定的處理邏輯，當一條日誌信息有幾十個甚至上百個字段的時候，在傳統的Flume處理下，收集到的日誌還是會有對應這麽多的字段，也不能對你想要的字段進行對應的處理。
2.2． 自定義攔截器
根據實際業務的需求，為了更好的滿足數據在應用層的處理，通過自定義Flume攔截器，過濾掉不需要的字段，並對指定字段加密處理，將源數據進行預處理。減少了數據的傳輸量，降低了存儲的開銷。
2.3． 功能實現
本技術方案核心包括二部分：
l 編寫java代碼，自定義攔截器
內容包括：

1. 定義一個類CustomParameterInterceptor實現Interceptor接口。
2. 在CustomParameterInterceptor類中定義變量，這些變量是需要到 Flume的配置文件中進行配置使用的。每一行字段間的分隔符(fields_separator)、通過分隔符分隔後，所需要列字段的下標（indexs）、多個下標使用的分隔符（indexs_separator)、多個下標使用的分隔符（indexs_separator)。
3. 添加CustomParameterInterceptor的有參構造方法。並對相應的變量進行處理。將配置文件中傳過來的unicode編碼進行轉換為字符串。
4. 寫具體的要處理的邏輯intercept()方法，一個是單個處理的，一個是批量處理。
5. 接口中定義了一個內部接口Builder，在configure方法中，進行一些參數配置。並給出，在flume的conf中沒配置一些參數時，給出其默認值。通過其builder方法，返回一個CustomParameterInterceptor對象。
6. 定義一個靜態類，類中封裝MD5加密方法
   
7. 通過以上步驟，自定義攔截器的代碼開發已完成，然後打包成jar， 放到Flume的根目錄下的lib中
   l 修改Flume的配置信息
   新增配置文件spool-interceptor-hdfs.conf，內容為：
   a1.channels = c1
   a1.sources = r1
   a1.sinks = s1
   #channel
   a1.channels.c1.type = memory
   a1.channels.c1.capacity=100000
   a1.channels.c1.transactionCapacity=50000
   #source
   a1.sources.r1.channels = c1
   a1.sources.r1.type = spooldir
   a1.sources.r1.spoolDir = /root/data/
   a1.sources.r1.batchSize= 50
   a1.sources.r1.inputCharset = UTF-8
   a1.sources.r1.interceptors =i1 i2
   a1.sources.r1.interceptors.i1.type =cn.itcast.interceptor.CustomParameterInterceptor$Builder
   a1.sources.r1.interceptors.i1.fields_separator=\u0009
   a1.sources.r1.interceptors.i1.indexs =0,1,3,5,6
   a1.sources.r1.interceptors.i1.indexs_separator =\u002c
   a1.sources.r1.interceptors.i1.encrypted_field_index =0
   a1.sources.r1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
   #sink
   a1.sinks.s1.channel = c1
   a1.sinks.s1.type = hdfs
   a1.sinks.s1.hdfs.path =hdfs://192.168.200.101:9000/flume/%Y%m%d
   a1.sinks.s1.hdfs.filePrefix = event
   a1.sinks.s1.hdfs.fileSuffix = .log
   a1.sinks.s1.hdfs.rollSize = 10485760
   a1.sinks.s1.hdfs.rollInterval =20
   a1.sinks.s1.hdfs.rollCount = 0
   a1.sinks.s1.hdfs.batchSize = 1500
   a1.sinks.s1.hdfs.round = true
   a1.sinks.s1.hdfs.roundUnit = minute
   a1.sinks.s1.hdfs.threadsPoolSize = 25
   a1.sinks.s1.hdfs.useLocalTimeStamp = true
   a1.sinks.s1.hdfs.minBlockReplicas = 1
   a1.sinks.s1.hdfs.fileType =DataStream
   a1.sinks.s1.hdfs.writeFormat = Text
   a1.sinks.s1.hdfs.callTimeout = 60000
   a1.sinks.s1.hdfs.idleTimeout =60
   啟動：
   bin/flume-ng agent -c conf -f conf/spool-interceptor-hdfs.conf -name a1 -Dflume.root.logger=DEBUG,console

Flume實戰案例