2.5.7 hadoop体系之离线计算-Flume日志收集系统-Flume案例分析（拦截器）

短命女 2022-12-19 06:06 140阅读 0赞

**目录**

1.案例场景分析

2.数据流程处理分析

3.实现

3.1 采集端配置文件开发

3.2 服务端配置文件开发

3.3 采集端 文件生成脚本

3.4 顺序启动服务

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

# 1.案例场景分析 #

A、B两台日志服务机器实时生产日志主要类型为access.log、nginx.log、web.log

现在要求：  
把A、B 机器中的access.log、nginx.log、web.log 采集汇总到C机器上然后统一收集到hdfs中。

但是在hdfs中要求的目录为：  
/source/logs/access/20180101/\*\*  
/source/logs/nginx/20180101/\*\*  
/source/logs/web/20180101/\*\*

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

#  2.数据流程处理分析 #

利用拦截器，定义key和value，判断数据来源于哪一个文件，使得hdfs可以正确放置数据。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

# 3.实现 #

服务器A对应的IP为 192.168.174.100  
服务器B对应的IP为 192.168.174.110  
服务器C对应的IP为 192.168.174.120

## 3.1 **采集端配置文件开发** ##

**node03与node02服务器**开发flume的配置文件

cd /export/services/apache-flume-1.8.0-bin/conf/  
vim 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  /export/services/taillogs/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 /export/services/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/services/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
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = node01
    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

## 3.2 **服务端配置文件开发** ##

在node01上面开发flume配置文件

cd /export/services/apache-flume-1.8.0-bin/conf/  
vim avro\_source\_hdfs\_sink.conf

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # 定义source
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 192.168.188.100
    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.188.100: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
    
    # 时间类型
    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

## 3.3 **采集端 文件生成脚本** ##

在**node03与node02**上面开发shell脚本，模拟数据生成

cd /export/services/shells  
vim server.sh

# !/bin/bash
    while true
    do
    date >> /export/services/taillogs/access.log;
    date >> /export/services/taillogs/web.log;
    date >> /export/services/taillogs/nginx.log;
    sleep 0.5;
    done

## 3.4 **顺序启动服务** ##

**node01启动flume实现数据收集**

cd /export/services/apache-flume-1.8.0-bin/
    bin/flume-ng agent -c conf -f conf/avro_source_hdfs_sink.conf -name a1 -Dflume.root.logger=INFO,console

**node03与node02启动flume实现数据监控**

cd /export/services/apache-flume-1.8.0-bin/
    bin/flume-ng agent -c conf -f conf/exec_source_avro_sink.conf -name a1 -Dflume.root.logger=INFO,console

**node03 与node02启动生成文件脚本**

cd /export/services/shells/
    sh server.sh

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221120/056b5fc048264dea9309fc5237b8da88.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221120/42514f68932b40f882759d91bd0df8a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020110516540413.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU=,size_16,color_FFFFFF,t_70