首页 > 代码库 > 通过flume-ng收集日志
通过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; } @Override public void close() { } @Override public void initialize() { } private String[] split(String s) { return s.split("\t", -1); //一定要这么写-1，否则如果最后以\t结尾而又是空字段的话， java会split出错误的长度 } @Override public 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; } } @Override public 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 = ""; @Override public void configure(Context context) { header = context.getString("platform", "DEFAULT_PLATFORM"); } @Override public Interceptor build() { return new PlatformInterceptor(header,preHour,finalName); } } }
再写flume执行所依赖的配置文件：
#agent1agent1.sources=source1agent1.sinks=sink1agent1.channels=channel1#source1agent1.sources.source1.type=spooldir#agent1.sources.source1.type=regex_extractoragent1.sources.source1.spoolDir=/hadoop/flume/flumelog #获取日志的目录agent1.sources.source1.channels=channel1#agent1.sources.source1.deletePolicy=immediate #立即删除.COMPELETED文件agent1.sources.source1.fileHeader=falseagent1.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=http://www.mamicode.com/a1>
编写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
如有错误之处，请指正，不胜感激。
通过flume-ng收集日志