阿里云-云小站(无限量代金券发放中)
【腾讯云】云服务器、云数据库、COS、CDN、短信等热卖云产品特惠抢购

使用flume+kafka+storm构建实时日志分析系统

179次阅读
没有评论

共计 4776 个字符,预计需要花费 12 分钟才能阅读完成。

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

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

1. flume 安装使用
    下载 flume 安装包 http://www.apache.org/dyn/closer.cgi/flume/1.5.2/apache-flume-1.5.2-bin.tar.gz
    解压 $ 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,配置文件如下:

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

# 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

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

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

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

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

# 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 目录下可以看到生成的日志文件。

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

 

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

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

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

        eventKey = headers.get(KEY_HDR);

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

        if (event == null) {
        break;
        }

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

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

        eventKey = headers.get(KEY_HDR);

        if (logger.isDebugEnabled()) {
        logger.debug(“{Event} ” + eventTopic + ” : ” + eventKey + ” : “
        + eventBody);
        logger.debug(“event #{}”, processedEvents);
        }

        // create a message and add to buffer
        ProducerData<String, String> data = new ProducerData<String, String>
        (eventTopic, eventBody);
        messageList.add(data);
            }
           
            // publish batch and commit.
    if (processedEvents > 0) {
    long startTime = System.nanoTime();
    long endTime = System.nanoTime();
    }
        }
       
        transaction.commit();
    } catch (Exception ex) {
        String errorMsg = “Failed to publish events”;
        logger.error(“Failed to publish events”, ex);
        status = Status.BACKOFF;
        if (transaction != null) {
          try {
            transaction.rollback();
          } catch (Exception e) {
            logger.error(“Transaction rollback failed”, e);
            throw Throwables.propagate(e);
          }
        }
        throw new EventDeliveryException(errorMsg, ex);
      } finally {
        if (transaction != null) {
          transaction.close();
        }
      }
   
    return status;
    下一步,修改 flume 配置文件,将其中 sink 部分的配置改成 kafka sink,如:
   

producer.sinks.r.type = org.apache.flume.sink.kafka.KafkaSink

producer.sinks.r.brokerList = bigdata-node00:9092
producer.sinks.r.requiredAcks = 1
producer.sinks.r.batchSize = 100
#producer.sinks.r.kafka.producer.type=async
#producer.sinks.r.kafka.customer.encoding=UTF-8
producer.sinks.r.topic = testFlume1
    type 指向 kafkasink 所在的完整路径
    下面的参数都是 kafka 的一系列参数,最重要的是 brokerList 和 topic 参数

现在重新启动 flume,就可以在 kafka 的对应 topic 下查看到对应的日志

分布式发布订阅消息系统 Kafka 架构设计 http://www.linuxidc.com/Linux/2013-11/92751.htm

Apache Kafka 代码实例 http://www.linuxidc.com/Linux/2013-11/92754.htm

Apache Kafka 教程笔记 http://www.linuxidc.com/Linux/2014-01/94682.htm

Apache kafka 原理与特性(0.8V)  http://www.linuxidc.com/Linux/2014-09/107388.htm

Kafka 部署与代码实例  http://www.linuxidc.com/Linux/2014-09/107387.htm

Kafka 介绍和集群环境搭建  http://www.linuxidc.com/Linux/2014-09/107382.htm

Kafka 的详细介绍:请点这里
Kafka 的下载地址:请点这里

本文永久更新链接地址:http://www.linuxidc.com/Linux/2016-03/129286.htm

正文完
星哥说事-微信公众号
post-qrcode
 0
星锅
版权声明:本站原创文章,由 星锅 于2022-01-21发表,共计4776字。
转载说明:除特殊说明外本站文章皆由CC-4.0协议发布,转载请注明出处。
【腾讯云】推广者专属福利,新客户无门槛领取总价值高达2860元代金券,每种代金券限量500张,先到先得。
阿里云-最新活动爆款每日限量供应
评论(没有评论)
验证码
【腾讯云】云服务器、云数据库、COS、CDN、短信等云产品特惠热卖中