寫的Log4j輸出到HDFS

Question

有沒有人試圖寫log4j的日誌文件直接到Hadoop分佈式文件系統？

如果是，請回復如何實現此目的。 我想我將不得不爲它創建一個Appender。

是這樣嗎？ 我的必要性是以特定的間隔將日誌寫入文件，並在稍後階段查詢該數據。

Answer 1

我建議使用Apache Flume來執行此任務。有Flume appender for Log4j。這樣，您可以將日誌發送到Flume，並寫入HDFS。這種方法的好處在於Flume與HDFS成爲單一的通信點。 Flume可以輕鬆添加新的數據源，而無需一次又一次編寫與HDFS交互的代碼。

Answer 2

標準log4j（1.x）不支持寫入HDFS。但幸運的是，log4j很容易擴展。我寫了一個HDFS FileAppender來寫入日誌到MapRFS（與Hadoop兼容）。文件名可以像「maprfs：///projects/example/root.log」。它適用於我們的項目。我提取代碼的appender部分並將其粘貼到下面。代碼片段可能無法運行。但是這會給你想法如何寫你的appender。實際上，你只需要擴展org.apache.log4j.AppenderSkeleton，並實現append（），close（），requireLayout（）。有關更多信息，還可以下載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; 
    } 

}