現(xiàn)狀分析

如何解決我們的數(shù)據(jù)從其他的server上移動(dòng)到Hadoop之上

腳本shell cp到Hadoop集群的機(jī)器上，然后使用hadoop fs -put命令傳到hadoop上【問(wèn)題：1.這種方法如何做監(jiān)控，2.文本數(shù)據(jù)的傳輸對(duì)于磁盤的開銷非常大 3. 必須要指定一個(gè)間隔的時(shí)間，比如每隔1分鐘拷貝一次，這樣時(shí)效性不好 4. 如何做容錯(cuò)和負(fù)載均衡】

使用Flume。容錯(cuò)、負(fù)載均衡、高延遲、壓縮在flume中都有很好的解決。只需要寫config就可以了

Flume概述

Flume is a distributed, reliable, and available service for efficiently collecting, aggregating, and moving large amounts of log data.

主要包括收集(collecting)、聚合(aggregating)、移動(dòng)(moving)功能。
也就是說(shuō)webserver(源端)可以通過(guò)Flume移動(dòng)到HDFS(目的端)中。

Flume架構(gòu)

source 收集

channel 聚集

sink 輸出

業(yè)界同類產(chǎn)品對(duì)比

(常用)Flume：Apache項(xiàng)目，采用java進(jìn)行開發(fā)
Scribe：Facebook項(xiàng)目，采用C/C++開發(fā)，負(fù)載均衡與容錯(cuò)不是很好。目前不再維護(hù)。
Chukwa：Yahoo/Apache項(xiàng)目，采用java開發(fā)，不再維護(hù)
Fluentd：與Flume類似，采用Ruby開發(fā)
(常用)Logstash：ELK（Elasticsearch+Logstash+Kibana）

安裝Flume

前置條件

java1.8以上

足夠內(nèi)存，供source，channel，sink使用

磁盤空間足夠

文件目錄權(quán)限

下載安裝Flume

下載Flume：下載CDH5.15.1版本，這里

解壓：tar -xvf flume-ng-1.6.0-cdh5.15.1.tar.gz -C ~/app/
目錄結(jié)構(gòu)：

添加環(huán)境變量，~/.bashrc 內(nèi)容如下：

# FLUME_HOME 1.6.0 FLUME_HOME=/home/iie4bu/app/apache-flume-1.6.0-cdh5.15.1-binPATH=$JAVA_HOME/bin:$HADOOP_HOME/bin:$HIVE_HOME/bin:$SPARK_HOME/bin:$FLUME_HOME/bin:$PATHexport PATH=$PATH

然后執(zhí)行source ~/.bashrc使得環(huán)境變量生效

配置Flume

在conf目錄下執(zhí)行cp flume-env.sh.template flume-env.sh
添加export JAVA_HOME=/home/iie4bu/app/jdk1.8.0_101
檢查運(yùn)行情況，在bin目錄下執(zhí)行：./flume-ng version

Flume實(shí)戰(zhàn)

需求1

從指定網(wǎng)絡(luò)端口采集數(shù)據(jù)輸出到控制臺(tái)。

配置

使用Flume的關(guān)鍵就是寫配置文件

A) 配置Source B) 配置Channel C) 配置Sink D) 把以上三個(gè)組件串起來(lái) a1: agent名稱 r1: source的名稱 k1: sink的名稱 c1: channel的名稱

在conf目錄下新建一個(gè)example.conf, 內(nèi)容如下：

# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1# Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444# Describe the sink a1.sinks.k1.type = logger# 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

啟動(dòng)agent

bin/flume-ng agent --name a1 --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/example.conf -Dflume.root.logger=INFO,console

其中：

--name: 表示agent的名稱

–conf: 表示 $FLUME_HOME/conf

–conf-file: 指定自己寫的配置文件

-D 表示JDK的一些參數(shù)。-Dflume.root.logger=INFO,console

使用nc進(jìn)行測(cè)試

重新開一個(gè)窗口，使用nc命令在本地測(cè)試44444端口

在Flume中可以看到輸出結(jié)果：

在輸出的日志中可以看到Event：
Event: { headers:{} body: 61 62 63 abc }
這個(gè)Event就是Flume中數(shù)據(jù)傳輸?shù)幕締卧?/p>

需求2

監(jiān)控一個(gè)文件實(shí)時(shí)采集新增的數(shù)據(jù)輸出到控制臺(tái)

配置
Agent選型：exec source + memory channel + logger sink
在$FLUME_HOME/conf目錄下新建exec-memory-logger.conf配置文件，內(nèi)容如下：

# 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 /home/iie4bu/data/hello.txt a1.sources.r1.shell = /bin/sh -c# Describe the sink a1.sinks.k1.type = logger# 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

啟動(dòng)agent

bin/flume-ng agent --name a1 --conf $FLUME_HOME/conf --conf-file /home/iie4bu/app/apache-flume-1.6.0-cdh5.15.1-bin/conf/exec-memory-logger.conf -Dflume.root.logger=INFO,console

這樣就實(shí)現(xiàn)了監(jiān)聽/home/iie4bu/data/hello.txt文件，并把里面的內(nèi)容輸出到控制臺(tái)。

需求3

將A服務(wù)器上的日志實(shí)時(shí)采集到B服務(wù)器。

Agent選型

機(jī)器A上的agent選型：exec source + memory channel + avro sink
機(jī)器B上的agent選型：avro source + memory channel + logger sink

配置Agent
新建exec-memory-avro.conf，內(nèi)容如下：

exec-memory-avro.sources = exec-source exec-memory-avro.sinks = avro-sink exec-memory-avro.channels = memory-channel# Describe/configure the source exec-memory-avro.sources.exec-source.type = exec exec-memory-avro.sources.exec-source.command = tail -F /home/iie4bu/data/hello.txt exec-memory-avro.sources.exec-source.shell = /bin/sh -c# Describe the sink exec-memory-avro.sinks.avro-sink.type = avro exec-memory-avro.sinks.avro-sink.hostname = localhost exec-memory-avro.sinks.avro-sink.port = 44444# Use a channel which buffers events in memory exec-memory-avro.channels.memory-channel.type = memory exec-memory-avro.channels.memory-channel.capacity = 1000 exec-memory-avro.channels.memory-channel.transactionCapacity = 100# Bind the source and sink to the channel exec-memory-avro.sources.exec-source.channels = memory-channel exec-memory-avro.sinks.avro-sink.channel = memory-channel

新建avro-memory-logger.conf，內(nèi)容如下：

avro-memory-logger.sources = avro-source avro-memory-logger.sinks = logger-sink avro-memory-logger.channels = memory-channel# Describe/configure the source avro-memory-logger.sources.avro-source.type = avro avro-memory-logger.sources.avro-source.bind = localhost avro-memory-logger.sources.avro-source.port = 44444# Describe the sink avro-memory-logger.sinks.logger-sink.type = logger# Use a channel which buffers events in memory avro-memory-logger.channels.memory-channel.type = memory avro-memory-logger.channels.memory-channel.capacity = 1000 avro-memory-logger.channels.memory-channel.transactionCapacity = 100# Bind the source and sink to the channel avro-memory-logger.sources.avro-source.channels = memory-channel avro-memory-logger.sinks.logger-sink.channel = memory-channel

啟動(dòng)agent

這里要注意啟動(dòng)順序。
首先啟動(dòng)avro-memory-logger.conf：bin/flume-ng agent --name avro-memory-logger --conf $FLUME_HOME/conf --conf-file /home/iie4bu/app/apache-flume-1.6.0-cdh5.15.1-bin/conf/avro-memory-logger.conf -Dflume.root.logger=INFO,console

然后再啟動(dòng)exec-memory-avro.conf：bin/flume-ng agent --name exec-memory-avro --conf $FLUME_HOME/conf --conf-file /home/iie4bu/app/apache-flume-1.6.0-cdh5.15.1-bin/conf/exec-memory-avro.conf -Dflume.root.logger=INFO,console

當(dāng)我們給/home/iie4bu/data/hello.txt文件添加內(nèi)容時(shí)，在avro-memory-logger.conf就會(huì)打印輸出響應(yīng)：

延時(shí)

這里兩個(gè)Agent之間會(huì)有一定的延時(shí)，因?yàn)閏hannel是基于內(nèi)存，有大小設(shè)置，到了一定的時(shí)間才會(huì)進(jìn)行相應(yīng)的操作。

總結(jié)日志收集過(guò)程

機(jī)器A上監(jiān)控一個(gè)文件，當(dāng)我們?cè)L問(wèn)主站時(shí)會(huì)有用戶行為日志記錄到access.log中

avro sink把新產(chǎn)生的日志輸出到對(duì)應(yīng)的機(jī)器B的hostname和port上

通過(guò)機(jī)器B上的avro source對(duì)應(yīng)的agent將我們的日志輸出到控制臺(tái)（Kafka）

總結(jié)

以上是生活随笔為你收集整理的Spark Streaming（二）Flume的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。