通过flume-ng收集日志

　　最近接到一个日志收集的需求，经过测试和修改，目前基本实现想要的功能，记录一下。

先说一下日志收集的需求，每隔1小时收集一次log日志，按照类别生成不同的lzo压缩文件，而且生成的日志要放在到前一个小时的目录中。拿到这个需求先想到使用flume来进行日志收集，再用Interceptor进行过滤，可以通过自带的RegexFilteringInterceptor进行过滤，不过要处理的日志列数太多，写成正则表达式的话比较麻烦。于是乎就干脆自己写一个Interceptor好了，也比较利于以后的扩展。

首先下载flume（老版本下还要通过zookeeper进行），在最新的flume-ng下可以不依赖zookeeper，我使用的是flume1.5.0。

下载完之后，先把hadoop-lzo包拷贝到$FLUME_HOME/lib中，或者添加到flume-env.sh脚本中FLUME_CLASSPATH。

　　如果使用memory channel时，遇到OOM的问题，还需要增加flume启动内存，具体做法是在flume-env.sh中扩大jvm。例如

　　JAVA_OPTS="-Xms1500m -Xmx2000m -Dcom.sun.management.jmxremote"

请注意：如果要使用flume-env.sh这种方式的话，要把flume-ng agent --conf conf 命令中引用的conf路径写全，网上有直接写成flume-ng agent --conf conf 这样的，是不起作用的。要写--conf conf的全路径。比如我的路径是/home/max/flume/conf/，那么就要写成

flume-ng agent --conf /home/max/flume/conf/

　　编写Intercepter

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Iterator;
import java.util.List;
import java.util.Map;import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;public class PlatformInterceptor implements Interceptor {private final String header;private String preHour;private String finalName;private String platform;private PlatformInterceptor(String header,String preHour,String finalName) {this.header = header;this.preHour = preHour;this.finalName = finalName;}@Overridepublic void close() {}@Overridepublic void initialize() {}private String[] split(String s) {return s.split("\t", -1); //一定要这么写-1，否则如果最后以\t结尾而又是空字段的话， java会split出错误的长度}@Overridepublic Event intercept(Event event) {String line = new String(event.getBody());Map<String, String> headers = event.getHeaders();String[] arrLine = split(line);if (arrLine.length != 20) {return null;} else {            headers.put("platform", arrLine[3]); //第三列为业务类别   headers.put("preHour", preHour);headers.put("finalName", finalName);return event;}}@Overridepublic List<Event> intercept(List<Event> events) {//此处生成前一个小时的字符串格式Calendar cal = Calendar.getInstance();cal.add(Calendar.HOUR_OF_DAY, -1);SimpleDateFormat formatter = new SimpleDateFormat("yyyy/MM/dd/HH");preHour = formatter.format(cal.getTime());      formatter = new SimpleDateFormat("yyyyMMddHH");finalName = formatter.format(cal.getTime());for (Iterator<Event> iterator = events.iterator(); iterator.hasNext();) {Event next = intercept(iterator.next());if (next == null) {iterator.remove();}}return events;}public static class Builder implements Interceptor.Builder {private String header = "platform";private String preHour = "";private String finalName = "";@Overridepublic void configure(Context context) {header = context.getString("platform", "DEFAULT_PLATFORM");            }@Overridepublic Interceptor build() {         return new PlatformInterceptor(header,preHour,finalName);}      }   }

再写flume执行所依赖的配置文件：

#agent1
agent1.sources=source1
agent1.sinks=sink1
agent1.channels=channel1#source1
agent1.sources.source1.type=spooldir
#agent1.sources.source1.type=regex_extractor
agent1.sources.source1.spoolDir=/hadoop/flume/flumelog  #获取日志的目录
agent1.sources.source1.channels=channel1
#agent1.sources.source1.deletePolicy=immediate  #立即删除.COMPELETED文件agent1.sources.source1.fileHeader=false
agent1.sources.source1.interceptors=platform#agent1.sources.source1.interceptors.a1.type=regex_extractor #正则interceptors
#agent1.sources.source1.interceptors.a1.regex=(\\S+)  (\\S+) # 此处为正则表达式的内容，有多少列写成多少列。（我是这种笨办法）
#agent1.sources.source1.interceptors.a1.serializers=s1 s2
#agent1.sources.source1.interceptors.a1.serializers.s1.name=key1
#agent1.sources.source1.interceptors.a1.serializers.s2.name=key2#agent1.sources.source1.interceptors.a1.platform=a2
#agent1.sources.source1.interceptors.a1.key=a1#agent1.sources.source1.interceptors.a2.platform=a2
#agent1.sources.source1.interceptors.a2.key=a2#agent1.sources.source1.interceptors.a1.value=a1
#agent1.sources.source1.interceptors.PlatformInterceptor2.type=PlatformInterceptor2$Builder#sink1
agent1.sinks.sink1.type=hdfs
agent1.sinks.sink1.channel=channel1
agent1.sinks.sink1.hdfs.path=/user/max/flume-input/%{preHour} #生成前一个小时目录
agent1.sinks.sink1.hdfs.fileType=CompressedStreamagent1.sinks.sink1.hdfs.rollInterval=0
#131072000   128M   67108864 64M
#393216000   128*3  压缩比大概是1:3
#1572864000  128*3*4
agent1.sinks.sink1.hdfs.rollSize=1572864000
agent1.sinks.sink1.hdfs.rollCount=0
agent1.sinks.sink1.hdfs.idleTimeout=1800 # 如果没有event可写了，要等待这个时间之后才会关闭channel，即把.tmp 文件名修改成最终的文件名agent1.sinks.sink1.hdfs.codeC=LzopCodecagent1.sinks.sink1.hdfs.filePrefix=boo_%{platform}_%{finalName}_40.seqagent1.sinks.sink1.hdfs.useLocalTimeStamp = true
agent1.sinks.sink1.hdfs.threadsPoolSize=30agent1.channels.channel1.type=memory
agent1.channels.channel1.capacity = 1000000
agent1.channels.channel1.transactionCapacity = 1000000

编写sh

flume-ng agent --conf /hadoop/flume/flume/conf -n agent1 -C /home/max/PlatformInterceptor-0.0.1-SNAPSHOT.jar -f /hadoop/flume/project/new -Dflume.root.logger=INFO,console

参考：

http://flume.apache.org/FlumeUserGuide.html

如有错误之处，请指正，不胜感激。

转载于:https://www.cnblogs.com/nb591/p/4168452.html

通过flume-ng收集日志相关推荐

基于Flume的美团日志收集系统(二)改进和优化
在<基于Flume的美团日志收集系统(一)架构和设计>中,我们详述了基于Flume的美团日志收集系统的架构设计,以及为什么做这样的设计.在本节中,我们将会讲述在实际部署和使用过程中遇到的问 ...
基于Flume的美团日志收集系统(一)架构和设计
背景美团的日志收集系统负责美团的所有业务日志的收集,并分别给Hadoop平台提供离线数据和Storm平台提供实时数据流.美团的日志收集系统基于Flume设计和搭建而成. <基于Flume的美团 ...
基于Flume的美团日志收集系统-----架构和设计
问题导读: 1.Flume-NG与Scribe对比,Flume-NG的优势在什么地方? 2.架构设计考虑需要考虑什么问题? 3.Agent死机该如何解决? 4.Collector死机是否会有影响? 5 ...
flume三种方式收集日志的案例
exec 监测某个单一的文件 # user_action_log_src 这个名字可以任意取 agent1.sources = user_action_log_src # memoryChannel ...
Flume NG 简介及配置实战
2019独角兽企业重金招聘Python工程师标准>>> Flume NG 简介及配置实战博客分类: 分布式计算 1.Flume 的一些核心概念: 1.1 数据流模型 1.2 高可靠 ...
nginx+flume网络流量日志实时数据分析实战
文章目录 nginx+flume网络流量日志实时数据分析实战网络流量日志数据分析-概述网络流量日志数据分析-数据处理流程网络流量日志数据分析-数据采集网站日志文件启动nginx服务器: 刷新 ...
Flume OG和Flume NG的区别
应用场景 Flume作为Hadoop中的日志采集工具,非常的好用,但是在安装Flume的时候,查阅很多资料,发现形形色色,有的说安装Flume很简单,有的说安装Flume很复杂,需要依赖zookeep ...
nginx日志切割并使用flume-ng收集日志
nginx的日志文件没有rotate功能.如果你不处理,日志文件将变得越来越大,还好我们可以写一个nginx日志切割脚本来自动切割日志文件. 第一步就是重命名日志文件,不用担心重命名后nginx找不到 ...
Filebeat 收集日志的那些事儿
女主宣言最近因为云原生日志收集的需要,我们打算使用Filebeat作为容器日志收集工具,并对其进行二次开发,因此笔者将谈谈 Filebeat 收集日志的那些事儿.本文不涉及过具体的源码分析,希望通过 ...

通过flume-ng收集日志

通过flume-ng收集日志相关推荐

最新文章

热门文章