将 Log4j 输出写入 HDFS答案

【问题标题】：Write Log4j output to HDFS将 Log4j 输出写入 HDFS
【发布时间】：2013-05-07 13:21:14
【问题描述】：

有没有人尝试将log4j日志文件直接写入Hadoop分布式文件系统？

如果是，请回复如何实现。我想我必须为它创建一个 Appender。

是这样吗？我的需要是以特定的时间间隔将日志写入文件并在稍后阶段查询该数据。

【问题讨论】：

两年后有没有人知道如何在不使用水槽的情况下做到这一点？在我的情况下，日志甚至不够长，不需要它

【解决方案1】：

我建议使用Apache Flume 来完成这项任务。有Flume appender for Log4j。这样，您将日志发送到 Flume，然后它写入 HDFS。这种方法的好处是 Flume 成为与 HDFS 的单点通信。 Flume 使添加新数据源变得容易，而无需编写一堆代码来一次又一次地与 HDFS 交互。

【讨论】：

【解决方案2】：

标准 log4j(1.x) 不支持写入 HDFS。但幸运的是，log4j 很容易扩展。我已经编写了一个 HDFS FileAppender 来将日志写入 MapRFS（与 Hadoop 兼容）。文件名可以类似于“maprfs:///projects/example/root.log”。它在我们的项目中运行良好。我提取代码的 appender 部分并将其粘贴在下面。代码 sn-ps 可能无法运行。但这会让你知道如何编写附加程序。实际上，你只需要扩展 org.apache.log4j.AppenderSkeleton，并实现 append()、close()、requiresLayout()。有关更多信息，您还可以下载 log4j 1.2.17 源代码并查看 AppenderSkeleton 是如何定义的，它将在那里为您提供所有信息。祝你好运！

注意：写入 HDFS 的另一种方法是将 HDFS 挂载到所有节点，这样您就可以像写入本地目录一样写入日志。也许这是实践中更好的方法。

import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.spi.LoggingEvent;
import org.apache.log4j.Layout;
import org.apache.hadoop.conf.Configuration;

import java.io.*;

public class HDFSFileAppender {
    private String filepath = null;
    private Layout layout = null;

    public HDFSFileAppender(String filePath, Layout layout){
        this.filepath = filePath;
        this.layout = layout;
    }

    @Override
    protected void append(LoggingEvent event) {
        String log = this.layout.format(event);
        try {
            InputStream logStream = new ByteArrayInputStream(log.getBytes());
            writeToFile(filepath, logStream, false);
            logStream.close();
        }catch (IOException e){
            System.err.println("Exception when append log to log file: " + e.getMessage());
        }
    }

    @Override
    public void close() {}

    @Override
    public boolean requiresLayout() {
        return true;
    }

    //here write to HDFS
    //filePathStr: the file path in MapR, like 'maprfs:///projects/aibot/1.log'
    private boolean writeToFile(String filePathStr, InputStream inputStream, boolean overwrite) throws IOException {
        boolean success = false;
        int bytesRead = -1;
        byte[] buffer = new byte[64 * 1024 * 1024];
        try {
            Configuration conf = new Configuration();
            org.apache.hadoop.fs.FileSystem fs = org.apache.hadoop.fs.FileSystem.get(conf);
            org.apache.hadoop.fs.Path filePath = new org.apache.hadoop.fs.Path(filePathStr);
            org.apache.hadoop.fs.FSDataOutputStream fsDataOutputStream = null;

            if(overwrite || !fs.exists(filePath)) {
                fsDataOutputStream = fs.create(filePath, overwrite, 512, 3, 64*1024*1024);
            }else{ //append to existing file.
                fsDataOutputStream = fs.append(filePath, 512);
            }

            while ((bytesRead = inputStream.read(buffer)) != -1) {
                fsDataOutputStream.write(buffer, 0, bytesRead);
            }

            fsDataOutputStream.close();
            success = true;
        } catch (IOException e) {
            throw e;
        }
        return success;
    }

}

【讨论】：