直截了当的解决办法应该是使用四个 HdfsBolt 和一个附加的“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());
    }
}

构建拓扑时，不同的 HdfsBolt 订阅不同的流：

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中配置为自己的文件。