Flume介紹

簡述

• Flume是一個分布式、可靠、和高可用的?量日志采集、聚合和傳輸的系統
• Flume可以采集檔案，socket資料包、檔案、檔案夾、kafka等各種形式源資料，又可以將采集到的資料(下沉sink)輸出到HDFS、hbase、hive、kafka等眾多外部存盤系統中
• 一般的采集需求，通過對flume的簡單配置即可實作
• Flume針對特殊場景也具備良好的自定義擴展能力，因此，flume可以適用于大部分的日常資料采集場景

運行機制

1.Flume分布式系統中最核心的角色是agent，flume采集系統就是由一個個agent所連接起來形成

2.每一個agent相當于一個資料傳遞員，內部有三個組件：

• Source：采集組件，用于跟資料源對接，以獲取資料
• Sink：下沉組件，用于往下一級agent傳遞資料或者往最終存盤系統傳遞資料
• Channel：傳輸通道組件，用于從source將資料傳遞到sink

Flume 結構圖

簡單結構

單個 Agent 采集資料

復雜結構

多級 Agent 之間串聯

Flume簡單示例

使用網路telent命令向一臺機?發送一些網路資料，然后通過flume采集網路埠資料

Flume 的安裝部署

下載解壓修改組態檔

下載地址： http://archive.apache.org/dist/flume/1.8.0/apache-flume-1.8.0-bin.tar.gz
Flume的安裝非常簡單，只需要解壓即可，當然，前提是已有hadoop環境，上傳安裝包到資料源所在節點上
cd /export/softwares/
tar -zxvf apache-flume-1.8.0-bin.tar.gz -C …/servers/
cd /export/servers/apache-flume-1.8.0-bin/conf
cp flume-env.sh.template flume-env.sh
vim flume-env.sh
export JAVA_HOME=/export/servers/jdk1.8.0_141

開發組態檔

根據資料采集的需求配置采集方案，描述在組態檔中(檔案名可任意自定義)，配置我們的網路收集的組態檔
在flume的conf目錄下新建一個組態檔（采集方案）
vim /export/servers/apache-flume-1.8.0-bin/conf/netcat-logger.conf

# 定義這個agent中各組件的名字 
a1.sources = r1 
a1.sinks = k1 
a1.channels = c1 

# 描述和配置source組件：r1 
a1.sources.r1.type = netcat 
a1.sources.r1.bind = 192.168.174.1 
a1.sources.r1.port = 44444 

# 描述和配置sink組件：k1 
a1.sinks.k1.type = logger 

# 描述和配置channel組件，此處使用是記憶體快取的方式 
a1.channels.c1.type = memory 
a1.channels.c1.capacity = 1000 
a1.channels.c1.transactionCapacity = 100 

# 描述和配置source channel sink之間的連接關系 
a1.sources.r1.channels = c1 
a1.sinks.k1.channel = c1

啟動組態檔

指定采集方案組態檔，在相應的節點上啟動flume agent，先用一個最簡單的例子來測驗一下程式環境是否正常，啟動agent去采集資料
bin/flume-ng agent -c conf -f conf/netcat-logger.conf -n a1 -Dflume.root.logger=INFO,console

• c conf 指定flume自身的組態檔所在目錄
• f conf/netcat-logger.con 指定我們所描述的采集方案
• n a1 指定我們這個agent的名字

安裝 Telnet 準備測驗

在機?上面安裝telnet客戶端，用于模擬資料的發送

yum -y install telnet
telnet node03 44444  #使用telnet模擬資料發送

采集示例

采集目錄到 HDFS

某服務?的某特定目錄下，會不斷產生新的檔案，每當有新檔案出現，就需要把檔案采集到HDFS中去

根據需求，首先定義以下3大要素

1. 資料源組件，即source ——監控檔案目錄 : spooldir
   • 監視一個目錄，只要目錄中出現新檔案，就會采集檔案中的內容
   • 采集完成的檔案，會被agent自動添加一個后綴：COMPLETED
   • 所監視的目錄中不允許重復出現相同檔案名的檔案
2. 下沉組件，即sink——HDFS檔案系統 : hdfs sink
3. 通道組件，即channel——可用file channel 也可以用記憶體channel

Flume 組態檔

cd /export/servers/apache-flume-1.8.0-bin/conf
mkdir -p /export/servers/dirfile
vim spooldir.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
##注意：不能往監控目中重復丟同名檔案
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /export/servers/dirfile
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = hdfs://node:8020/spooldir/files/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的檔案型別，默認是Sequencefile，可用DataStream，則為普通文本
a1.sinks.k1.hdfs.fileType = DataStream

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Channel引數解釋：
capacity：默認該通道中最大的可以存盤的event數量
trasactionCapacity：每次最大可以從source中拿到或者送到sink中的event數量
keep-alive：event添加到通道中或者移出的允許時間

啟動 Flume

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

上傳檔案到指定目錄

將不同的檔案上傳到下面目錄里面去，注意檔案不能重名
cd /export/servers/dirfile

采集檔案到 HDFS

業務系統使用log4j生成的日志，日志內容不斷增加，需要把追加到日志檔案中的資料實時采集到hdfs

根據需求，首先定義以下3大要素：

• 采集源，即source——監控檔案內容更新 : exec ‘tail -F file’
• 下沉目標，即sink——HDFS檔案系統 : hdfs sink
• Source和sink之間的傳遞通道——channel，可用file channel 也可以用 記憶體channel

定義 Flume 組態檔

cd /export/servers/apache-flume-1.8.0-bin/conf
vim tail-file.conf

agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

# Describe/configure tail -F source1
agent1.sources.source1.type = exec
agent1.sources.source1.command = tail -F /export/servers/taillogs/access_log
agent1.sources.source1.channels = channel1

# Describe sink1
agent1.sinks.sink1.type = hdfs
#a1.sinks.k1.channel = c1
agent1.sinks.sink1.hdfs.path = hdfs://node01:8020/weblog/flume-collection/%y-%m-%d/%H-%
agent1.sinks.sink1.hdfs.filePrefix = access_log
agent1.sinks.sink1.hdfs.maxOpenFiles = 5000
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text
agent1.sinks.sink1.hdfs.round = true
agent1.sinks.sink1.hdfs.roundValue = 10
agent1.sinks.sink1.hdfs.roundUnit = minute
agent1.sinks.sink1.hdfs.useLocalTimeStamp = true

# Use a channel which buffers events in memory
agent1.channels.channel1.type = memory
agent1.channels.channel1.keep-alive = 120
agent1.channels.channel1.capacity = 500000
agent1.channels.channel1.transactionCapacity = 600

# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1

啟動 Flume

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -c conf -f conf/tail-file.conf -n agent1 -Dflume.root.logger=INFO,console

開發 Shell 腳本定時追加檔案內容

mkdir -p /export/servers/shells/
cd /export/servers/shells/
vim tail-file.sh

#!/bin/bash
while true
do
 date >> /export/servers/taillogs/access_log;
 sleep 0.5;
done

啟動腳本

# 創建檔案夾
mkdir -p /export/servers/taillogs
# 啟動腳本
sh /export/servers/shells/tail-file.sh

Agent級聯

第一個agent負責收集檔案當中的資料，通過網路發送到第二個agent當中去
第二個agent負責接收第一個agent發送的資料，并將資料保存到hdfs上面去

Node2安裝Flume

將第一臺機?上面解壓后的flume檔案夾拷貝到第二臺機?上面去
cd /export/servers
scp -r apache-flume-1.8.0-bin/ node2:$PWD

Node2配置Flume

在node2機?配置flume
cd /export/servers/ apache-flume-1.8.0-bin/conf
vim tail-avro-avro-logger.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/servers/taillogs/access_log
a1.sources.r1.channels = c1

# Describe the sink
##sink端的avro是一個資料發送者
a1.sinks = k1
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 192.168.174.120
a1.sinks.k1.port = 4141
a1.sinks.k1.batch-size = 10

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

開發腳本向檔案中寫入資料

直接將第一臺下面的腳本和資料拷貝到node2即可，第一臺機?上執行以下命令
cd /export/servers
scp -r shells/ taillogs/ node2:$PWD

第一臺Flume組態檔

在第一臺機?上開發flume的組態檔
cd /export/servers/apache-flume-1.8.0-bin/conf
vim avro-hdfs.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
##source中的avro組件是一個接收者服務
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 192.168.xxx.xxx
a1.sources.r1.port = 6666

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://node01:8020/av /%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的檔案型別，默認是Sequencefile，可用DataStream，則為普通文本
a1.sinks.k1.hdfs.fileType = DataStream

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

順序啟動

第一臺機?啟動flume行程
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -c conf -f conf/avro-hdfs.conf -n a1 -Dflume.root.logger=INFO,console

node2機?啟動flume行程
cd /export/servers/apache-flume-1.8.0-bin/
bin/flume-ng agent -c conf -f conf/tail-avro-avro-logger.conf -n a1 -Dflume.root.logger

node2機?啟shell腳本生成檔案
cd /export/servers/shells
sh tail-file.sh

高可用方案

示例搭建一個高可用的Flume NG集群

角色分配

Flume的Agent和Collector分布如下表所示：

圖中所示，Agent1資料分別流入到Collector1和Collector2，Flume NG本身提供了Failover機制，可以
自動切換和恢復，有3個產生日志服務?分布在不同的機房，要把所有的日志都收集到一個
集群中存盤，

Node01安裝和配置

將flume安裝包以及檔案生產的兩個目錄拷貝到node01機?上面去

node02機?執行以下命令

cd /export/servers
scp -r apache-flume-1.8.0-bin/ node01:$PWD
scp -r shells/ taillogs/ node01:$PWD

node01機?配置agent的組態檔

cd /export/servers/apache-flume-1.8.0-bin/conf
vim agent.conf

#agent1 name
agent1.channels = c1
agent1.sources = r1
agent1.sinks = k1 k2

##set gruop
agent1.sinkgroups = g1

agent1.sources.r1.channels = c1
agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F /export/servers/taillogs/access_log

##set channel
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 1000
agent1.channels.c1.transactionCapacity = 100

## set sink1
agent1.sinks.k1.channel = c1
agent1.sinks.k1.type = avro
agent1.sinks.k1.hostname = node02
agent1.sinks.k1.port = 52020

## set sink2
agent1.sinks.k2.channel = c1
agent1.sinks.k2.type = avro
agent1.sinks.k2.hostname = node03
agent1.sinks.k2.port = 52020

##set sink group
agent1.sinkgroups.g1.sinks = k1 k2

##set failover
agent1.sinkgroups.g1.processor.type = failover
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
agent1.sinkgroups.g1.processor.maxpenalty = 10000

Node02 與 Node03 配置 FlumeCollection

node02機?修改組態檔

cd /export/servers/apache-flume-1.8.0-bin/conf
vim collector.conf

#set Agent name
a1.sources = r1
a1.channels = c1
a1.sinks = k1

##set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

## other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = node02
a1.sources.r1.port = 52020
a1.sources.r1.channels = c1

##set sink to hdfs
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path= hdfs://node01:8020/flume/failover/
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

node03機?修改組態檔

cd /export/servers/apache-flume-1.8.0-bin/conf
vim collector.conf

#set Agent name
a1.sources = r1
a1.channels = c1
a1.sinks = k1

##set channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

## other node,nna to nns
a1.sources.r1.type = avro
a1.sources.r1.bind = node03
a1.sources.r1.port = 52020
a1.sources.r1.channels = c1

##set sink to hdfs
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path= hdfs://node01:8020/flume/failover/
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=10
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

順序啟動

node03機?上面啟動flume
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/collector.conf -Dflume.root.logger=DEBUG,console

node02機?上面啟動flume
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/collector.conf -Dflume.root.logger=DEBUG,console

node01機?上面啟動flume
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -n agent1 -c conf -f conf/agent.conf -Dflume.root.logger=DEBUG,console

node01機?啟動檔案產生腳本
cd /export/servers/shells
sh tail-file.sh

Flume 的負載均衡

負載均衡是用于解決一臺機?(一個行程)無法解決所有請求而產生的一種演算法，Load balancing Sink Processor 能夠實作 load balance 功能，如下圖Agent1 是一個路由節點，負責將 Channel 暫存的Event 均衡到對應的多個 Sink組件上，而每個 Sink 組件分別連接到一個獨立的 Agent 上，示例配置，如下所示：

通過三臺機?來進行模擬flume的負載均衡
三臺機?規劃如下：
node01：采集資料，發送到node02和node03機?上去
node02：接收node01的部分資料
node03：接收node01的部分資料

步驟一：開發node01服務?的flume配置

node01服務?配置：

cd /export/servers/apache-flume-1.8.0-bin/conf
vim load_banlancer_client.conf

# agent name
<p class="mume-header " id="agent-name"></p>
a1.channels = c1
a1.sources = r1
a1.sinks = k1 k2

# set gruop
<p class="mume-header " id="set-gruop"></p>
a1.sinkgroups = g1

# set channel
<p class="mume-header " id="set-channel"></p>
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sources.r1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/servers/taillogs/access_log

# set sink1
<p class="mume-header " id="set-sink1"></p>
a1.sinks.k1.channel = c1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = node02
a1.sinks.k1.port = 52020

# set sink2
<p class="mume-header " id="set-sink2"></p>
a1.sinks.k2.channel = c1
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = node03
a1.sinks.k2.port = 52020

# set sink group
<p class="mume-header " id="set-sink-group"></p>
a1.sinkgroups.g1.sinks = k1 k2

# set failover
<p class="mume-header " id="set-failover"></p>
a1.sinkgroups.g1.processor.type = load_balance
a1.sinkgroups.g1.processor.backoff = true
a1.sinkgroups.g1.processor.selector = round_robin
a1.sinkgroups.g1.processor.selector.maxTimeOut=10000

步驟二：開發node02服務?的flume配置

node02服務?配置：

cd /export/servers/apache-flume-1.8.0-bin/conf
vim load_banlancer_server.conf

# Name the components on this agent
<p class="mume-header " id="name-the-components-on-this-agent"></p>
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
<p class="mume-header " id="describeconfigure-the-source"></p>
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = node02
a1.sources.r1.port = 52020

# Describe the sink
<p class="mume-header " id="describe-the-sink"></p>
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
<p class="mume-header " id="use-a-channel-which-buffers-events-in-memory"></p>
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
<p class="mume-header " id="bind-the-source-and-sink-to-the-channel"></p>
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

步驟三：開發node03服務?flume配置

node03服務?配置

cd /export/servers/apache-flume-1.8.0-bin/conf
vim load_banlancer_server.conf

# Name the components on this agent
<p class="mume-header " id="name-the-components-on-this-agent-1"></p>
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
<p class="mume-header " id="describeconfigure-the-source-1"></p>
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = node03
a1.sources.r1.port = 52020

# Describe the sink
<p class="mume-header " id="describe-the-sink-1"></p>
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
<p class="mume-header " id="use-a-channel-which-buffers-events-in-memory-1"></p>
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
<p class="mume-header " id="bind-the-source-and-sink-to-the-channel-1"></p>
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

步驟四：準備啟動flume服務

啟動node03的flume服務
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/load_banlancer_server.conf -Dflume.root.logger

啟動node02的flume服務
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/load_banlancer_server.conf -Dflume.root.logger

啟動node01的flume服務
cd /export/servers/apache-flume-1.8.0-bin
bin/flume-ng agent -n a1 -c conf -f conf/load_banlancer_client.conf -Dflume.root.logger

步驟五：node01服務?運行腳本產生資料

cd /export/servers/shells
sh tail-file.sh

Flume案例

A、B兩臺日志服務機?實時生產日志主要型別為access.log、nginx.log、web.log，現在需要把A、B 機?中的access.log、nginx.log、web.log 采集匯總到C機?上然后統一收集到hdfs中，但是在hdfs中要求的目錄為：/source/logs/access/20210101/** /source/logs/nginx/20210101/** /source/logs/web/20210101/**

服務?A對應的IP為 192.168.174.1
服務?B對應的IP為 192.168.174.2
服務?C對應的IP為 192.168.174.3

采集端組態檔開發

node01與node02服務?開發flume的組態檔

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim exec_source_avro_sink.conf

# Name the components on this agent
<p class="mume-header " id="name-the-components-on-this-agent-2"></p>
a1.sources = r1 r2 r3
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
<p class="mume-header " id="describeconfigure-the-source-2"></p>
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/servers/taillogs/access.log
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
## static攔截?的功能就是往采集到的資料的header中插入自己定義的key-value對
<p class="mume-header " id="static攔截?的功能就是往采集到的資料的header中插入自己定義的key-value"></p>
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 /export/servers/taillogs/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 /export/servers/taillogs/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
<p class="mume-header " id="describe-the-sink-2"></p>
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = node03
a1.sinks.k1.port = 41414

# Use a channel which buffers events in memory
<p class="mume-header " id="use-a-channel-which-buffers-events-in-memory-2"></p>
a1.channels.c1.type = memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity = 10000

# Bind the source and sink to the channel
<p class="mume-header " id="bind-the-source-and-sink-to-the-channel-2"></p>
a1.sources.r1.channels = c1
a1.sources.r2.channels = c1
a1.sources.r3.channels = c1
a1.sinks.k1.channel = c1

服務端組態檔開發

在node03上面開發flume組態檔

cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin/conf
vim avro_source_hdfs_sink.conf

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

# 定義source
<p class="mume-header " id="定義source"></p>
a1.sources.r1.type = avro
a1.sources.r1.bind = 192.168.174.3
a1.sources.r1.port =41414
# 添加時間攔截?
<p class="mume-header " id="添加時間攔截?"></p>
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

# 定義channels
<p class="mume-header " id="定義channels"></p>
a1.channels.c1.type = memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity = 10000

# 定義sink
<p class="mume-header " id="定義sink"></p>
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path=hdfs://192.168.174.1:8020/source/logs/%{type}/%Y%m%d
a1.sinks.k1.hdfs.filePrefix =events
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
# 時間型別
<p class="mume-header " id="時間型別"></p>
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# 生成的檔案不按條數生成
<p class="mume-header " id="生成的檔案不按條數生成"></p>
a1.sinks.k1.hdfs.rollCount = 0
# 生成的檔案按時間生成
<p class="mume-header " id="生成的檔案按時間生成"></p>
a1.sinks.k1.hdfs.rollInterval = 30
# 生成的檔案按大小生成
<p class="mume-header " id="生成的檔案按大小生成"></p>
a1.sinks.k1.hdfs.rollSize = 10485760
# 批量寫入hdfs的個數
<p class="mume-header " id="批量寫入hdfs的個數"></p>
a1.sinks.k1.hdfs.batchSize = 10000
# flume操作hdfs的執行緒數（包括新建，寫入等）
<p class="mume-header " id="flume操作hdfs的執行緒數包括新建寫入等"></p>
a1.sinks.k1.hdfs.threadsPoolSize=10
# 操作hdfs超時時間
<p class="mume-header " id="操作hdfs超時時間"></p>
a1.sinks.k1.hdfs.callTimeout=30000

# 組裝source、channel、sink
<p class="mume-header " id="組裝source-channel-sink"></p>
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

采集端檔案生成腳本

在node01與node02上面開發shell腳本，模擬資料生成

cd /export/servers/shells
vim server.sh

# !/bin/bash
<p class="mume-header " id="binbash"></p>
while true
do
	date >> /export/servers/taillogs/access.log;
	date >> /export/servers/taillogs/web.log;
	date >> /export/servers/taillogs/nginx.log;
	sleep 0.5;
done

順序啟動服務

node03啟動flume實作資料收集
cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -c conf -f conf/avro_source_hdfs_sink.conf -name a1 -Dflume.root.logger=DEBUG,console

node01與node02啟動flume實作資料監控
cd /export/servers/apache-flume-1.6.0-cdh5.14.0-bin
bin/flume-ng agent -c conf -f conf/exec_source_avro_sink.conf -name a1 -Dflume.root.logger=DEBUG,console

node01與node02啟動生成檔案腳本
cd /export/servers/shells
sh server.sh

tips：解決中文亂碼