Flume实战案例

1. 日志的采集和汇总
1.1. 案例场景
A、B两台日志服务机器实时生产日志主要类型为access.log、nginx.log、web.log
现在要求:
把A、B 机器中的access.log、nginx.log、web.log 采集汇总到C机器上然后统一收集到hdfs中。
但是在hdfs中要求的目录为:
/source/logs/access/20160101/
/source/logs/nginx/20160101/

/source/logs/web/20160101/**
1.2. 场景分析

1.3. 数据流程处理分析

1.4. 功能实现
① 在服务器A和服务器B上
创建配置文件 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 /root/data/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 /root/data/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 /root/data/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 = 192.168.200.101
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
② 在服务器C上创建配置文件 avro_source_hdfs_sink.conf 文件内容为
#定义agent名, source、channel、sink的名称
a1.sources = r1
a1.sinks = k1
a1.channels = c1
#定义source
a1.sources.r1.type = avro
a1.sources.r1.bind = mini2
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.200.101:9000/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
③ 配置完成之后,在服务器A和B上的/root/data有数据文件access.log、nginx.log、web.log。先启动服务器C上的flume,启动命令
在flume安装目录下执行 :
bin/flume-ng agent -c conf -f conf/avro_source_hdfs_sink.conf -name a1 -Dflume.root.logger=DEBUG,console
然后在启动服务器上的A和B,启动命令
在flume安装目录下执行 :
bin/flume-ng agent -c conf -f conf/exec_source_avro_sink.conf -name a1 -Dflume.root.logger=DEBUG,console

2. Flume自定义拦截器(了解)
2.1. 案例背景介绍
Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统,Flume支持在日志系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可定制)的能力。Flume有各种自带的拦截器,比如:TimestampInterceptor、HostInterceptor、RegexExtractorInterceptor等,通过使用不同的拦截器,实现不同的功能。但是以上的这些拦截器,不能改变原有日志数据的内容或者对日志信息添加一定的处理逻辑,当一条日志信息有几十个甚至上百个字段的时候,在传统的Flume处理下,收集到的日志还是会有对应这么多的字段,也不能对你想要的字段进行对应的处理。
2.2. 自定义拦截器
根据实际业务的需求,为了更好的满足数据在应用层的处理,通过自定义Flume拦截器,过滤掉不需要的字段,并对指定字段加密处理,将源数据进行预处理。减少了数据的传输量,降低了存储的开销。
2.3. 功能实现
本技术方案核心包括二部分:
l 编写java代码,自定义拦截器
内容包括:

  1. 定义一个类CustomParameterInterceptor实现Interceptor接口。
  2. 在CustomParameterInterceptor类中定义变量,这些变量是需要到 Flume的配置文件中进行配置使用的。每一行字段间的分隔符(fields_separator)、通过分隔符分隔后,所需要列字段的下标(indexs)、多个下标使用的分隔符(indexs_separator)、多个下标使用的分隔符(indexs_separator)。
  3. 添加CustomParameterInterceptor的有参构造方法。并对相应的变量进行处理。将配置文件中传过来的unicode编码进行转换为字符串。
  4. 写具体的要处理的逻辑intercept()方法,一个是单个处理的,一个是批量处理。
  5. 接口中定义了一个内部接口Builder,在configure方法中,进行一些参数配置。并给出,在flume的conf中没配置一些参数时,给出其默认值。通过其builder方法,返回一个CustomParameterInterceptor对象。
  6. 定义一个静态类,类中封装MD5加密方法
  7. 通过以上步骤,自定义拦截器的代码开发已完成,然后打包成jar, 放到Flume的根目录下的lib中
    l 修改Flume的配置信息
    新增配置文件spool-interceptor-hdfs.conf,内容为:
    a1.channels = c1
    a1.sources = r1
    a1.sinks = s1
    #channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity=100000
    a1.channels.c1.transactionCapacity=50000
    #source
    a1.sources.r1.channels = c1
    a1.sources.r1.type = spooldir
    a1.sources.r1.spoolDir = /root/data/
    a1.sources.r1.batchSize= 50
    a1.sources.r1.inputCharset = UTF-8
    a1.sources.r1.interceptors =i1 i2
    a1.sources.r1.interceptors.i1.type =cn.itcast.interceptor.CustomParameterInterceptor$Builder
    a1.sources.r1.interceptors.i1.fields_separator=\u0009
    a1.sources.r1.interceptors.i1.indexs =0,1,3,5,6
    a1.sources.r1.interceptors.i1.indexs_separator =\u002c
    a1.sources.r1.interceptors.i1.encrypted_field_index =0
    a1.sources.r1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
    #sink
    a1.sinks.s1.channel = c1
    a1.sinks.s1.type = hdfs
    a1.sinks.s1.hdfs.path =hdfs://192.168.200.101:9000/flume/%Y%m%d
    a1.sinks.s1.hdfs.filePrefix = event
    a1.sinks.s1.hdfs.fileSuffix = .log
    a1.sinks.s1.hdfs.rollSize = 10485760
    a1.sinks.s1.hdfs.rollInterval =20
    a1.sinks.s1.hdfs.rollCount = 0
    a1.sinks.s1.hdfs.batchSize = 1500
    a1.sinks.s1.hdfs.round = true
    a1.sinks.s1.hdfs.roundUnit = minute
    a1.sinks.s1.hdfs.threadsPoolSize = 25
    a1.sinks.s1.hdfs.useLocalTimeStamp = true
    a1.sinks.s1.hdfs.minBlockReplicas = 1
    a1.sinks.s1.hdfs.fileType =DataStream
    a1.sinks.s1.hdfs.writeFormat = Text
    a1.sinks.s1.hdfs.callTimeout = 60000
    a1.sinks.s1.hdfs.idleTimeout =60
    启动:
    bin/flume-ng agent -c conf -f conf/spool-interceptor-hdfs.conf -name a1 -Dflume.root.logger=DEBUG,console

原文地址:http://blog.51cto.com/13587708/2176119

时间: 2024-11-14 18:57:36

Flume实战案例的相关文章

Spark2.0从入门到精通:Scala编程、大数据开发、上百个实战案例、内核源码深度剖析视频教程

38套大数据,云计算,架构,数据分析师,Hadoop,Spark,Storm,Kafka,人工智能,机器学习,深度学习,项目实战视频教程 视频课程包含: 38套大数据和人工智能精品高级课包含:大数据,云计算,架构,数据挖掘实战,实时推荐系统实战,电视收视率项目实战,实时流统计项目实战,离线电商分析项目实战,Spark大型项目实战用户分析,智能客户系统项目实战,Linux基础,Hadoop,Spark,Storm,Docker,Mapreduce,Kafka,Flume,OpenStack,Hiv

运维实战案例之“Argument list too long”错误与解决方法

作为一名运维人员来说,这个错误并不陌生,在执行rm.cp.mv等命令时,如果要操作的文件数很多,可能会使用通配符批量处理大量文件,这时就可能会出现"Argument list too long"这个问题了. 1.错误现象 这是一台Mysql数据库服务器,在系统中运行了很多定时任务,今天通过crontab命令又添加了一个计划任务,退出时发生了如下报错: #crontab -e 编辑完成后,保存退出,就出现下面如下图所示错误: 2.解决思路 根据上面报错的提示信息,基本判定是磁盘空间满了,

Linux系统shell脚本编程——生产实战案例

Linux系统shell脚本编程--生产实战案例     在日常的生产环境中,可能会遇到需要批量检查内网目前在线的主机IP地址有哪些,还可能需要检查这些在线的主机哪些端口是开放状态,因此依靠手工来检查是可以实现,但比较费时费力,所以需要结合shell脚本来实现批量检查的功能,那么今天就来做个小小的实验. 1.开发脚本前准备 一般大家都知道,测试主机是否在线,常用的命令无非就是ping.nmap,因此,首先找一个地址来测试下ping命令的效果 [[email protected] scripts]

运维实战案例之文件已删除但空间不释放问题解析

1.错误现象 运维的监控系统发来通知,报告一台服务器空间满了,登陆服务器查看,根分区确实没有空间了,如下图所示: 这里首先说明一下服务器的一些删除策略,由于Linux没有回收站功能,我们的线上服务器所有要删除的文件都会首先移动到系统/tmp目录下,然后定期清除/tmp目录下的数据.这个策略本身没有问题,但是通过检查发现这台服务器的系统分区中并没有单独划分/tmp分区,这样/tmp下的数据其实是占用了根分区的空间.既然找到了问题,那么删除/tmp目录下一些大数据即可,执行如下命令,检查/tmp下最

Flume 实战(1) -- 初体验

前言: Flume-ng是数据收集/聚合/传输的组件, Flume-ng抛弃了Flume OG原本繁重的zookeeper和Master, Collector, 其整体的架构更加的简洁和明了. 其基础组件就Agent进程, 内部又可以细分为Source, Channel, Sink三个组件, Source是数据的输入源, channel作为消息的管道, 而sink是作为数据流的输出, Source可以配置多个channel, sink和channel一一对应. *) 初体验Flume-ng 以C

《Web渗透技术及实战案例解析》pdf

下载地址:网盘下载 内容简介 编辑 本书从Web渗透的专业角度,结合网络安全中的实际案例,图文并茂地再现Web渗透的精彩过程.本书共分7章,由浅入深地介绍和分析了目前网络流行的Web渗透攻击方法和手段,并结合作者多年的网络安全实践经验给出了相对应的安全防范措施,对一些经典案例还给出了经验总结和技巧,通过阅读本书可以快速掌握目前Web渗透的主流技术.本书最大的特色就是实用和实战性强,思维灵活.内容主要包括Web渗透必备技术.Google黑客技术.文件上传渗透技术.SQL注入.高级渗透技术.0day

Storm容错机制Acker详解和实战案例

Storm中有个特殊的Executor叫acker,他们负责跟踪spout发出的每一个Tuple的Tuple树.当acker发现一个Tuple树已经处理完成了,它会告诉框架回调Spout的ack(),否则回调Spout的fail(). Acker的跟踪算法是Storm的主要突破之一,对任意大的一个Tuple树,它只需要恒定的20字节就可以进行跟踪. 我们期望的是,如果某个Tuple被Bolt执行失败了,则Spout端可以重新发送该Tuple.但很遗憾的是,框架不会自动重新发送,需要我们自己手工编

Flume 实战(2)--Flume-ng-sdk源码分析

具体参考: 官方用户手册和开发指南 http://flume.apache.org/FlumeDeveloperGuide.html *) 定位和简单例子 1). Flume-ng-sdk是用于编写往flume agent发送数据的client sdk2). 简单示例 RpcClient client = null; try { client = RpcClientFactory.getDefaultInstance("127.0.0.1", 41414); Event event =

_00017 Flume的体系结构介绍以及Flume入门案例(往HDFS上传数据)

博文作者:妳那伊抹微笑 个性签名:世界上最遥远的距离不是天涯,也不是海角,而是我站在妳的面前,妳却感觉不到我的存在 技术方向:hadoop,数据分析与挖掘 转载声明:可以转载, 但必须以超链接形式标明文章原始出处和作者信息及版权声明,谢谢合作! qq交流群:214293307  (期待与你一起学习,共同进步) # 学习前言 想学习一下Flume,网上找了好多文章基本上都说的很简单,只有一半什么的,简直就是坑爹,饿顿时怒火就上来了,学个东西真不容易,然后自己耐心的把这些零零碎碎的东西整理整理,各种