Flume实战案例 02

男娘i 2022-02-28 12:24 249阅读 0赞

# 1. Flume的安装部署 #

## 1.1 需求 ##

案例：使用网络telent命令向一台机器发送一些网络数据，然后通过flume采集网络端口数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]

## 1.2 分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]

## 1.3 实现 ##

*  下载安装包，解压，修改配置文件  
    Flume的安装非常简单，只需要解压即可，当然，前提是已有hadoop环境  
    上传安装包到数据源所在节点上

tar -zxvf flume-ng-1.6.0-cdh5.14.0.tar.gz -C /export/servers/
    cd  /export/servers/apache-flume-1.6.0-cdh5.14.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.6.0-cdh5.14.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.52.120
    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的名字

*  安装telent 准备测试  
    在node02机器上面安装telnet客户端，用于模拟数据的发送

yum -y install telnet
    telnet  node03  44444   # 使用telnet模拟数据发送

![在这里插入图片描述][20190325195309794.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 2]

# 2. 采集案例 #

## 2.1 采集目录到hdfs中 ##

### 2.1.1 需求 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 3]

### 2.1.2 分析 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 4]

*  数据源组件，即source ——监控文件目录 : spooldir  
    spooldir特性：  
    1、监视一个目录，只要目录中出现新文件，就会采集文件中的内容  
    2、采集完成的文件，会被agent自动添加一个后缀：COMPLETED  
    3、所监视的目录中不允许重复出现相同文件名的文件
 *  下沉组件，即sink——HDFS文件系统 : hdfs sink
 *  通道组件，即channel——可用file channel 也可以用内存channel

### 2.1.3 Flume配置文件开发 ###

*  配置文件

cd  /export/servers/apache-flume-1.6.0-cdh5.14.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 = 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

roll size  10000条  往hdfs上面穿
    batch size   100条  100次
    roll count  5M

*  启动flume

bin/flume-ng agent -c ./conf -f ./conf/spooldir.conf -n a1 -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

### 2.1.4 Flume有没有做监控 ###

*  第一个：数据源，数据一直在增多，但是没有改变后缀名。说白了就是.complete 的文件没有增多
    
    第二个：数据的目的地，数据的目的地没有增多
    
    依据以上两个现象：就可以判断flume有没有在干活儿。判断flume没有干活儿了？？？杀掉flume的进程，然后将文件夹下面的文件全部移走，然后重新启动flume即可

可以通过shell脚本来实现  
需要自己通过shell脚本，或者java程序，定时的检测flume是否正常工作

*  flum多长时间滚动一次？？？？问你的flume的配置，多长时间上传一次到hdfs上面去
    
    500Gb数据 1小时42Gb数据 每隔10分钟产生768M数据
    
    第一个控制文件的大小1Gb 第二个 每隔10分钟滚动一次 第三个：每隔200event 滚动一次
    
    一天大概所有的数据在50-100Gb基本上你们可以驾驭 一般一到2个人可以搞定

flume还可以与kafka整合：做实时的数据的处理。不存在时间间隔的问题，将数据越快放入kafka越好

## 2.2 采集文件到hdfs中 ##

### 2.2.1 需求分析 ###

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

### 2.2.2 分析 ###

根据需求，首先定义以下3大要素  
采集源，即source——监控文件内容更新 : exec ‘tail -F file’  
下沉目标，即sink——HDFS文件系统 : hdfs sink  
Source和sink之间的传递通道——channel，可用file channel 也可以用 内存channel

### 2.2.3 实现 ###

*  定义flume配置文集

cd /export/servers/apache-flume-1.6.0-cdh5.14.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
    
    #configure host for source
    #agent1.sources.source1.interceptors = i1
    #agent1.sources.source1.interceptors.i1.type = host
    #agent1.sources.source1.interceptors.i1.hostHeader = hostname
    
    # 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.rollSize = 102400
    agent1.sinks.sink1.hdfs.rollCount = 1000000
    agent1.sinks.sink1.hdfs.rollInterval = 60
    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

## 2.3 两个agent级联 ##

### 2.3.1 需求 ###

第一个agent负责收集文件当中的数据，通过网络发送到第二个agent当中去，第二个agent负责接收第一个agent发送的数据，并将数据保存到hdfs上面去

### 2.3.1 分析 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 5]

### 2.3.2 实现 ###

*  node02安装flume  
    将node03机器上面解压后的flume文件夹拷贝到node02机器上面去

cd  /export/servers
    scp -r apache-flume-1.6.0-cdh5.14.0-bin/ node02:$PWD

![在这里插入图片描述][20190326091218331.png]

*  node02配置flume配置文集  
    在node02机器配置我们的flume

cd /export/servers/apache-flume-1.6.0-cdh5.14.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.52.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

*  node02开发脚本文件往文件写入数据  
    直接将node03下面的脚本和数据拷贝到node02即可，node03机器上执行以下命令

cd  /export/servers
    scp -r shells/ taillogs/ node02:$PWD

*  node03开发flume配置文件

cd /export/servers/apache-flume-1.6.0-cdh5.14.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.52.120
    a1.sources.r1.port = 4141
    # Describe the sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://node01:8020/avro/hdfs/%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

*  顺序启动  
    node03机器启动flume进程

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

node0机器启动flume进程

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

node02机器启动shell脚本生成文件

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

# 3. 更多source和sink组件 #

[Flume支持众多的source和sink类型，详细手册可参考官方文档][Flume_source_sink]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: /images/20220228/8e0cc469e5be43abbe98394fd78dede2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]: /images/20220228/ecfee27d3f92429aa77e6d0c95ea5dac.png
[20190325195309794.png]: /images/20220228/f941c8ad34424c54b332915a99fa3f6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 2]: /images/20220228/503fcd92e9744eca810b08579e885758.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 3]: /images/20220228/73ce4e618fea45fe9db88db68e7a3553.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 4]: /images/20220228/16acb4a5658c4a83b19b1021714714c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 5]: /images/20220228/4193ccad49bd4caa856c3fde8e98700d.png
[20190326091218331.png]: /images/20220228/92e636159467438fba0555981a569c6d.png
[Flume_source_sink]: http://archive.cloudera.com/cdh5/cdh/5/flume-ng-1.6.0-cdh5.14.0/FlumeUserGuide.html