2.5.3 hadoop体系之离线计算-Flume日志收集系统-Flume实战（采集目录+采集日志）

桃扇骨 2022-12-19 03:37 158阅读 0赞

**目录**

1.spooldir采集目录到HDFS

1.1 思路分析

1.2 Flume 配置文件

1.3 启动Flume

1.4 上传文件到指定目录

2.exec采集文件到 HDFS

2.1 思路分析

2.2 定义Flume配置文件

2.3 启动 Flume

2.4 开发 Shell 脚本定时追加文件内容

2.5 启动脚本

--------------------

# 1.spooldir采集目录到HDFS #

我们服务器上在某一个文件夹下面可能有很多文件，而且这个文件是不断变化的，我们可以利用flume采集该文件目录下面文件的变化。当该文件夹下面有新文件产生的时候，我们可以采集这个数据。

![20201105105237703.png][]

实现步骤：

（1）写一个配置文件xxx.conf：配置source、channel，sink

（2）启动Flume的Agent实例

（3）运行测试

## 1.1 思路分析 ##

某服务器的某特定目录下，会不断产生新的文件，每当有新文件出现，就需要把文件采集到HDFS中去，根据需求，首先定义以下3大要素：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

## 1.2 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://node01:8020/spooldir/files/%y-%m-%d/%H%M/
    a1.sinks.k1.hdfs.filePrefix = eventsa1.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添加到通道中或者移出的允许时间

sink参数解释

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

## 1.3 启动Flume ##

![20201105110641616.png][]

## 1.4 上传文件到指定目录 ##

将不同的文件上传到下面目录里面去，注意文件不能重名

cd /export/servers/dirfile

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

文件自动加上，completed表示文件已经被flume采集过了，介于我们配置的触发条件，生成了两个hdfs文件。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3][]

# 2.exec采集文件到 HDFS #

需求 比如业务系统**使用log4j生成的日志**，**日志内容不断增加**，需要把**追加到日志文件中的数据**实时采集到 hdfs

## 2.1 思路分析 ##

根据需求，首先定义以下3大要素

![20201105110004706.png][]

## 2.2 定义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-%M
    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

## 2.3 启动 Flume ##

创建文件夹和文件

![20201105112010220.png][]

启动flume

![20201105112054128.png][]

## 2.4 开发 Shell 脚本定时追加文件内容 ##

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

## 2.5 启动脚本 ##

\# 启动脚本 **sh /export/servers/shells/tail-file.sh**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4][]

[20201105105237703.png]: /images/20221120/942c2a535eee4c2fa851466487dfa658.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221120/fd81f27fc77845f18c1ecda5378bc0a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221120/ca056dbdaf254b3982857cc6757000a2.png
[20201105110641616.png]: /images/20221120/82248bb423a646449bb8c93b2e959879.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221120/16d56e6aa6d74516a055e1f79212db79.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3]: /images/20221120/bcb1d50ef9ea450fb964112330a2339b.png
[20201105110004706.png]: /images/20221120/24dcf2e7e14c4636833b3db00b5f7879.png
[20201105112010220.png]: /images/20221120/dd834cd953534407962b1cd0efd4b4a6.png
[20201105112054128.png]: /images/20221120/bed530be70a14d9e8b69fa5fc4e73300.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4]: /images/20221120/03bbbd444fae48b1b9d36e368e4ba903.png