根据日志级别写入HDFS答案

【问题标题】：Write into HDFS based on log level根据日志级别写入HDFS
【发布时间】：2016-02-17 07:01:53
【问题描述】：

我正在使用 HDFS bolt 将日志消息写入 HDFS。现在所有具有不同日志级别的日志消息，如 WARN、DEBUG、INFO、ERROR 都写入同一个文件。如何在一个文件中写入 INFO 消息，在另一个文件中写入 WARN 消息，在 HDFS 的另一个文件中单独写入 ERROR 消息？我正在使用 log4j Kafka 附加程序。基本上我需要根据INFO，ERROR和DEBUG等日志级别将日志消息写入不同的文件。

【问题讨论】：

标签： hadoop log4j hdfs apache-kafka apache-storm

【解决方案1】：

直接的解决方案应该是使用四个HdfsBolts 和一个额外的“LogSplitBolt”，它发出四个输出流（每个日志级别一个）：

public class LogSplitBolt {
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        Fields schema = new Fields(...);
        declarer.declareStream("debug", schema);
        declarer.declareStream("error", schema);
        declarer.declareStream("warnings", schema);
        declarer.declareStream("info", schema);
   }

   public void execute(Tuple input) {
       String logLevel = input.getXXX(...); // get log level
       // use logLevel as output streamId
       collector.emit(logLevel, new Values(input.getValues());
    }
}

当您构建拓扑时，不同的HdfsBolts 订阅不同的流：

builder.addBolt("splitter", new LogSplitBolt());
builder.addbolt("writeDebug", new HdfsBolt(...)).localOfShuffle("debug", "splitter");
builder.addbolt("writeError", new HdfsBolt(...)).localOfShuffle("error", "splitter");
builder.addbolt("writeWarning", new HdfsBolt(...)).localOfShuffle("warning", "splitter");
builder.addbolt("writeInfo", new HdfsBolt(...)).localOfShuffle("info", "splitter");

当然，每个HdfsBolt 都在HDFS 中配置为自己的文件。

【讨论】：

是的，或者你可以只写一个螺栓来完成所有这些。