通过flume-ng收集日志

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

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);
		}		
		
	}	
	
}

#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=CompressedStream

agent1.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=LzopCodec

agent1.sinks.sink1.hdfs.filePrefix=boo_%{platform}_%{finalName}_40.seq


agent1.sinks.sink1.hdfs.useLocalTimeStamp = true 
agent1.sinks.sink1.hdfs.threadsPoolSize=30


agent1.channels.channel1.type=memory
agent1.channels.channel1.capacity = 1000000
agent1.channels.channel1.transactionCapacity = 1000000