菜单

storm构建实时日志分析系统,Flume对Nginx群集日志收集方案

2019年8月6日 - 4166m金沙
storm构建实时日志分析系统,Flume对Nginx群集日志收集方案

本文只会涉及flume和kafka的结合,kafka和storm的结合可以参考其他文章。

Flume是一个分布式、可靠、和高可用的海量日志聚合的系统,支持在系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可定制)的能力。

Kafka-Storm 集成部署
http://www.linuxidc.com/Linux/2016-03/129063.htm

flume的基本概念
如下图

  1. flume安装使用
       
    下载flume安装包
        解压$ tar -xzvf apache-flume-1.5.2-bin.tar.gz -C /opt/flume
        flume配置文件放在conf文件目录下,执行文件放在bin文件目录下。
        1)配置flume
       
    进入conf目录将flume-conf.properties.template拷贝一份,并命名为自己需要的名字
        $ cp flume-conf.properties.template flume.conf
        修改flume.conf的内容,我们使用file
    sink来接收channel中的数据,channel采用memory channel,source采用exec
    source,配置文件如下:

图片 1

agent.sources = seqGenSrc
agent.channels = memoryChannel
agent.sinks = loggerSink

每个agent都具有三个元素,source、channel、sink。
source:数据流的源。产生event。
channel:可以理解成数据流的管道。传递event
sink :  数据流的终点。消耗event
注:source可以上一节点的sink,sink可以指定为下一节点的source。比较常见的场景如下图

# For each one of the sources, the type is defined
agent.sources.seqGenSrc.type = exec
agent.sources.seqGenSrc.command = tail -F /data/mongodata/mongo.log
#agent.sources.seqGenSrc.bind = 172.168.49.130

图片 2

# The channel can be defined as follows.
agent.sources.seqGenSrc.channels = memoryChannel

为以收集日志,并做实时的集中存储,元素相应类型如下
1.source : client端使用exec类型,通过tail –F
产生event。server端使用avro类型,用于接收client端发出的event
2.channel : 使用file类型。(测试期使用了mem类似)

# Each sink’s type must be defined
agent.sinks.loggerSink.type = file_roll
agent.sinks.loggerSink.sink.directory = /data/flume

  1. sink:
    client端使用avro类型,传递给server端。server端使用file_roll类型,指定相应目录储存日志。最终方案会使用hdfs

#Specify the channel the sink should use
agent.sinks.loggerSink.channel = memoryChannel

图片 3

# Each channel’s type is defined.
agent.channels.memoryChannel.type = memory

flume具体配置如下:
1、将flume 解压到 /usr/local/flume下
2、agent端配置

# Other config values specific to each type of channel(sink or
source)
# can be defined as well
# In this case, it specifies the capacity of the memory channel
agent.channels.memoryChannel.capacity = 1000
agent.channels.memory4log.transactionCapacity = 100
    2)运行flume agent
    切换到bin目录下,运行一下命令:
    $ ./flume-ng agent –conf ../conf -f ../conf/flume.conf –n agent
-Dflume.root.logger=INFO,console
    在/data/flume目录下可以看到生成的日志文件。

flume允许对环境资源使用做设置,需要修改配置,/PREFIX/conf/flume-env.sh 
可以通过实际情况进行调整
JAVA_OPTS=”-Xms100m -Xmx200m -Dcom.sun.management.jmxremote”
#此处PREFIX 代表 /usr/local/flume

  1. 结合kafka
        由于flume1.5.2没有kafka sink,所以需要自己开发kafka sink
        可以参考flume 1.6里面的kafka
    sink,但是要注意使用的kafka版本,由于有些kafka api不兼容的
        这里只提供核心代码,process()内容。

配置(/PREFIX/conf/flume-client.properties)
# Name the components on this agent
a1.sources = r1
a1.channels = c1
a1.sinks = k1

 

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -n 0 -F /space/nginxlogs/access.web.log
a1.sources.r1.channels = c1

Sink.Status status = Status.READY;

 

    Channel ch = getChannel();
    Transaction transaction = null;
    Event event = null;
    String eventTopic = null;
    String eventKey = null;
   
    try {
        transaction = ch.getTransaction();
        transaction.begin();
        messageList.clear();
       
        if (type.equals(“sync”)) {
            event = ch.take();

# Describe/configure the channels (后面有memory channel配置方案)
a1.channels.c1.type = file
a1.channels.c1.keep-alive = 10
a1.channels.c1.write-timeout = 10
a1.channels.c1.checkpointDir =
/space/guang_mobileapi_flume/checkpoint
a1.channels.c1.useDualCheckpoints = true
a1.channels.c1.backupCheckpointDir = 
/space/guang_mobileapi_flume/backupcheckpoint
a1.channels.c1.dataDirs = /space/guang_mobileapi_flume/data

    if (event != null) {
        byte[] tempBody = event.getBody();
        String eventBody = new String(tempBody,”UTF-8″);
        Map<String, String> headers = event.getHeaders();

# Describe/configure the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.10.35
a1.sinks.k1.port = 44444
a1.sinks.k1.channel = c1

        if ((eventTopic = headers.get(TOPIC_HDR)) == null) {
        eventTopic = topic;
        }

(# Describe/configure the channels  (次方案可以替换前面的file
channel配置方案)
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
) 

        eventKey = headers.get(KEY_HDR);

server配置:

        if (logger.isDebugEnabled()) {
        logger.debug(“{Event} ” + eventTopic + ” : ” + eventKey + ” :

        + eventBody);
        }
       
        ProducerData<String, Message> data = new
ProducerData<String, Message>
        (eventTopic, new Message(tempBody));
       
        long startTime = System.nanoTime();
        logger.debug(eventTopic+”++++”+eventBody);
        producer.send(data);
        long endTime = System.nanoTime();
    }
        } else {
            long processedEvents = 0;
            for (; processedEvents < batchSize; processedEvents += 1)
{
                event = ch.take();

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

相关文章

发表评论

电子邮件地址不会被公开。 必填项已用*标注

网站地图xml地图