log采集

需求及选型

昨天在做简单日志采集分析,使用技术如下

  1. filebeat:做日志的收集
  2. kafka: 日志中转,缓冲及解耦作用
  3. storm: 做日志实时分析

后续会再加上其它,如

  • elasticsearch : 做搜索
  • hbase : 存储

在此调研只使用最上面的三项。

kafka

其安装不做介绍。只需做简单配置,然后创建一个topic,命令如:

kafka-topics.sh --create --zookeeper 192.168.101.176:2181 --replication-factor 1 --partitions 1 --topic demolog1

filebeat

安装不做介绍。

修改filebeat.yml,注释掉其默认的elasticsearch的配置,修改要输出的log文件

在最后面增加输入到kafka的信息,如:

output.kafka:
  enabled: true
  hosts: ["192.168.101.176:9092"]
  topic: demolog1

在此遇到一个问题:配置好ip,但发现不能找到kafka,其内部都是使用的域名,应该为:先通过ip查到域名,然后使用域名去查找,则这时找不到相应的主机了,这时只需要
/etc/hosts里面配置好相应的域名即可

storm

引用最新的kafka,及其client,最新的使用的是storm-kafka-client,网上已有的内容基本都是针对老版本kafka的,老版本使用的是storm-kafka,需要使用的pom如下:

<dependencies>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>1.1.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-kafka-client</artifactId>
        <version>1.1.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.12</artifactId>
        <version>0.11.0.1</version>
        <exclusions>
            <exclusion>
                <artifactId>log4j</artifactId>
                <groupId>log4j</groupId>
            </exclusion>
            <exclusion>
                <artifactId>slf4j-log4j12</artifactId>
                <groupId>org.slf4j</groupId>
            </exclusion>
        </exclusions>
    </dependency>
</dependencies>

这里需要把kafka中依赖的log4j去除,否则会报错,如果使用的是老版本的storm-kafka,则启动不了,是因为引入即有log4j,又有log4j2导致的。如果使用的storm-kafka-client,则会有以下
警告。

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/E:/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.8.2/log4j-slf4j-impl-2.8.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/E:/repository/org/slf4j/slf4j-log4j12/1.7.25/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]

此外还有一个问题是,虽然配置什么都正确,但是不能获取到kafka消息,最后发现是没有在host里面配置域名,虽然代码使用的是ip,但被解析后,再使用就不行了,增加host后解决。其原因应该和上面一样.

在此只从kafka中获得消息,不需要把消息写入到kafka,因此使用KafkaSpout

topology如下:

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

public class DemoTopology {
    private static TopologyBuilder builder = new TopologyBuilder();

    public static void main(String[] args) {
        Config config = new Config();
        config.setDebug(true);
        config.setNumWorkers(1);

        KafkaSpoutConfig kafkaSpoutConfig = KafkaSpoutConfig.builder("192.168.101.176:9092",
                "demolog1")
                .setProp("client.id","hello")
                .setProp("group.id", "storm")
                .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
                .build();

        builder.setSpout("kafkaSpolt", new KafkaSpout<>(kafkaSpoutConfig), 1);
        builder.setBolt("demoBolt", new DemoBolt(), 2).shuffleGrouping("kafkaSpolt");
        if (args != null && args.length > 0) {
            try {
                config.setNumWorkers(1);
                StormSubmitter.submitTopology(args[0], config,builder.createTopology());
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else {
            config.setMaxTaskParallelism(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("demo", config, builder.createTopology());
            Utils.sleep(1000000000);
            cluster.killTopology("demo");
            cluster.shutdown();
        }
    }
}

则这时就可以在自己的Bolt中获取到消息了。

参考文档:http://storm.apache.org/releases/1.1.1/storm-kafka-client.html

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 135,083评论 19 139
  • 概述 监控预警平台, eagle + eye (鹰眼)的合体词, 寓意可以快速发现问题, 并及时作出响应,Eagl...
    Kungfu猫熊阅读 7,479评论 0 52
  • Spring Boot 参考指南 介绍 转载自:https://www.gitbook.com/book/qbgb...
    毛宇鹏阅读 46,989评论 6 342
  • kafka的定义:是一个分布式消息系统,由LinkedIn使用Scala编写,用作LinkedIn的活动流(Act...
    时待吾阅读 5,375评论 1 15
  • 本人陆陆续续接触了ELK的1.4,2.0,2.4,5.0,5.2版本,可以说前面使用当中一直没有太多感触,最近使用...
    三杯水Plus阅读 4,141评论 0 12