java - 将 Log4j 输出写入 HDFS

Question

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

如果是，请回复如何实现这一点。我想我将不得不为它创建一个 Appender。

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

Answer 1

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

Answer 2

标准 log4j(1.x) 不支持写入 HDFS。但幸运的是，log4j 很容易扩展。我已经编写了一个 HDFS FileAppender 来将日志写入 MapRFS（与 Hadoop 兼容）。文件名可以类似于“maprfs:///projects/example/root.log”。它在我们的项目中运行良好。我提取代码的 appender 部分并将其粘贴在下面。代码片段可能无法运行。但这会让您知道如何编写附加程序。实际上，你只需要扩展 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;
    }

}