flume的部署及简单测试
Flume 的一些核心概念：组件功能Agent使用JVM 运行Flume。每台机器运行一个agent，但是可以在一个agent中包含多个sources和sinks。Client生产数据，运行在一个独立的线程。Source从Client收集数据，传递给Channel。Sink从Channel收集数据，运行在一个独立线程。Channel连接 sources 和 sinks ，这个有点像一个队列。Events可以是日志记录、 avro 对象等。配置flume
第一步：
下载最新的flume-ng
http://flume.apache.org/download.html
第二步：
配置conf
创建：flume-source-conf.properties
source：
#Describesource
source.sources.src1.type = exec //类型
source.sources.src1.command = tail-n +0 -F /home/hduser/logs/debug.log //监控的日志文件
#Describe channel //配置channel
source.channels.ch1.type = file
source.channels.ch1.checkpointDir = tmp/channels/ch1/checkpoint
source.channels.ch1.dataDirs = tmp/channels/ch1/data
#Describe sink //配置sink
source.sinks.sink2.type = file_roll
source.sinks.sink2.sink.directory=store
source.sinks.sink2.sink.serializer = text
source.sinks.sink2.sink.serializer.appendNewline = true
source.sinks.sink2.sink.rollInterval = 43200
//配置变量：
vim ~/.bash_profile
#export FLUME_HOME
export FLUME_HOME=/home/hduser/softsare/flume-1.6.0
export PATH=$PATH:$FLUME_HOME/sbin
. ~/.bash_profile //一定执行者一步让变量生效
启动命令：flume strart source
停止命令：flume stop source
//到此简单的flume配置完成，，可以用下面的日志生成程序产生日志。
package main;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;
import thread.DebugLogThread;
import thread.ErrorLogThread;
import thread.InfoLogThread;
import util.JFile;
public class LogFactory {
static Loggerlogger = Logger.getLogger(LogFactory.class);
/**
* 绾跨▼瀹瑰櫒锛屼綔涓虹瓑寰呮墍鏈夌嚎绋嬮�鍑轰箣鐢ㄣ�
*/
private static ArrayList<Thread> threads = new ArrayList<Thread>();
/**
* 鍚姩鏍稿績鍔熻兘鐩稿叧鐨勬墍鏈夋敞鍐岀嚎绋�
*/
public static void setpid() {
String processName = java.lang.management.ManagementFactory.getRuntimeMXBean().getName();
String processID = processName.substring(0,processName.indexOf('@'));
try {
JFile.write("pid.log", processID);
logger.debug("pid="+processID);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
/**
* 鍚姩鏍稿績鍔熻兘鐩稿叧鐨勬墍鏈夋敞鍐岀嚎绋�
*/
public static void start() {
// 鍚姩绾跨▼
for (int i = 0; i < 2; i++)
{
DebugLogThread log = new DebugLogThread();
Thread thread = new Thread(log);
thread.start();
threads.add(thread);
}//for
// 鍚姩绾跨▼
for (int i = 0; i < 2; i++)
{
ErrorLogThread log = new ErrorLogThread();
Thread thread = new Thread(log);
thread.start();
threads.add(thread);
}//for
// 鍚姩绾跨▼
for (int i = 0; i < 2; i++)
{
InfoLogThread log = new InfoLogThread();
Thread thread = new Thread(log);
thread.start();
threads.add(thread);
}//for
/**
* 绛夊緟鎵�湁绾跨▼閫�嚭
*/
for (int i = 0; i < threads.size(); i++)
{
try
{
// threads.get(i).join();
Thread.sleep(1);
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws ClassNotFoundException {
/**
* 閰嶇疆log4j鐨勯厤缃枃浠讹紝娉ㄦ剰杩欐槸xml鏍煎紡銆�
*/
DOMConfigurator.configure("etc/log4j.xml" );
/**
* 鍙傛暟鍒ゆ柇
*/
if (args.length < 1) {
System.out.println("args error, args="+Arrays.toString(args));
System.exit(-1);
}
System.out.println("args="+Arrays.toString(args));
/**
* 璁板綍杩涚▼鍙�
*/
setpid();
start();
logger.debug("process exit.");
}
}
//使用tail -f XXXX（你监控的文件）来跟踪日志生成情况；
页:
[1]