資料採集之Web端匯入日誌檔案到Hadoop HDFS
前言
接上一篇《資料採集之Web端匯入DB資料到Hadoop HDFS》,這一篇簡單的記錄一下如何在Web端控制匯入日誌檔案到HDFS中,主要用到的技術就是Flume了。網上大多數教程都是寫的配置檔案,但是現在有需求要通過web來動態自動化的進行操作,也就是說要通過Java程式碼的形式來實現,而不是直接修改伺服器上的配置檔案。所以只能想想其他其他方式了。
環境
- OS Debian 8.7
- Hadoop 2.6.5
- SpringBoot 1.5.1.RELEASE
- Flume 1.7.0
Flume簡介
Flume is a distributed, reliable, and available service for efficiently collecting, aggregating, and moving large amounts of log data. It has a simple and flexible architecture based on streaming data flows. It is robust and fault tolerant with tunable reliability mechanisms and many failover and recovery mechanisms. It uses a simple extensible data model that allows for online analytic application.
Flume是一個分散式、高可靠、高可用的日誌資料採集、聚合和傳輸服務。
它有一個簡單靈活基於資料流的結構。它也有可靠的和容錯的機制,並且可用進行許多故障轉移和恢復機制。 它使用一個簡單的可擴充套件資料模型,允許線上分析應用程式。
翻譯的不太好,簡單的理解就是一個日誌資料採集工具吧。
專案依賴
先看下pom.xml檔案。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.infosys.flume</groupId>
<artifactId>flume</artifactId>
<version>1.0-SNAPSHOT</version>
<name>flume</name>
<packaging>jar</packaging>
<parent>
<groupId >org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<hadoop.version>2.6.5</hadoop.version>
<flume.version>1.7.0</flume.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
<version>3.1.0</version>
<scope>provided</scope>
</dependency>
<!-- Flume -->
<dependency>
<groupId>org.apache.flume.flume-ng-sinks</groupId>
<artifactId>flume-hdfs-sink</artifactId>
<version>${flume.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- Hadoop -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- Test -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.mrunit</groupId>
<artifactId>mrunit</artifactId>
<version>1.1.0</version>
<classifier>hadoop2</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minicluster</artifactId>
<version>${hadoop.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>${project.artifactId}</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<configuration>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<outputDirectory>${basedir}</outputDirectory>
</configuration>
</plugin>
</plugins>
</build>
</project>
其中SpringBoot和Hadoop不多說,主要是org.apache.flume.flume-ng-sinks這個依賴不能少。它提供了很多方便配置flume相關的API。
核心Flume
這裡僅作為演示,真實專案肯定不能這樣的。因為網上包括官網都很少提供這種API的配置形式的,大都是直接改配置檔案。在這裡我們只需要將Flume下載好,然後使用預設配置即可。其他的配置在程式碼中編寫。
為了偷懶,跟上篇一樣,都寫在一個java類中了。。
package com.infosys.flume;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
import org.apache.flume.sink.hdfs.HDFSEventSink;
import org.apache.flume.source.SpoolDirectorySource;
import org.apache.flume.source.SpoolDirectorySourceConfigurationConstants;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
/**
* 描述:
* 公司:infosys(印孚瑟斯技術有限公司)
* 作者:luhaoyuan <[email protected]>
* 版本:1.0
* 建立日期:2017/3/15.
*/
@RestController
public class FlumeController {
private Channel memChannel = new MemoryChannel();
private HDFSEventSink hdfsSink;
@PostMapping(value = "/log2Hdfs")
public String logToHdfs() throws EventDeliveryException {
configSources();
configChannels();
configSinks();
// 儲存event到HDFS
hdfsSink.process();
return "SUCCESS";
}
/**
* 配置通道
*/
private void configChannels() {
Context channelContext = new Context();
memChannel.setName("ch1");
Configurables.configure(memChannel, channelContext);
memChannel.start();
}
/**
* 配置目的地,這裡就是匯入到HDFS
*/
private void configSinks() {
Context sinkContext = new Context();
sinkContext.put("hdfs.path", HdfsSinkConstants.HDFS_PATH);
sinkContext.put("hdfs.fileType", HdfsSinkConstants.HDFS_FILE_TYPE);
sinkContext.put("hdfs.filePrefix", "%{fileName}"); // 將原來的檔名(包括路徑)作為字首
sinkContext.put("hdfs.round", "true");
sinkContext.put("hdfs.rollInterval", "30"); // 隔30秒就將事件寫入HDFS
sinkContext.put("hdfs.useLocalTimeStamp", "true"); // 使用時間戳配置
hdfsSink = new HDFSEventSink();
hdfsSink.setName("myhdfs");
hdfsSink.configure(sinkContext);
hdfsSink.setChannel(memChannel);
hdfsSink.start();
}
/**
* 配置日誌源,指定一個目錄
*/
private void configSources() {
Context context = new Context();
SpoolDirectorySource spoolDirSource = new SpoolDirectorySource();
spoolDirSource.setName("nginx");
List<Channel> channels = new ArrayList<>();
channels.add(memChannel);
ChannelSelector rcs = new ReplicatingChannelSelector();
rcs.setChannels(channels);
spoolDirSource.setChannelProcessor(new ChannelProcessor(rcs));
// 日誌存放路徑
File file = new File("/var/log/nginx");
context.put(SpoolDirectorySourceConfigurationConstants.SPOOL_DIRECTORY, file.getAbsolutePath());
// 將檔名(包括路徑)新增到頭資訊,方便寫入的時候獲取
context.put(SpoolDirectorySourceConfigurationConstants.FILENAME_HEADER, "true");
context.put(SpoolDirectorySourceConfigurationConstants.FILENAME_HEADER_KEY, "fileName");
Configurables.configure(spoolDirSource, context);
spoolDirSource.start();
}
}
其實基本上也跟改配置檔案類似,除了用Context,也可以用JDK的Properties好像。其中配置大都可用通過前端表單的形式傳遞過來。
常量
就幾個常量,懶的抽了。
package com.infosys.flume;
/**
* 描述:
* 公司:infosys(印孚瑟斯技術有限公司)
* 作者:luhaoyuan <[email protected]>
* 版本:1.0
* 建立日期:2017/3/15.
*/
public final class HdfsSinkConstants {
public static final String HDFS_PATH = "hdfs://e5:9000/flume/nginx/%y-%m-%d"; //目錄按照年月日進行儲存
public static final String HDFS_FILE_TYPE = "DataStream"; // 表示檔案型別,不會被壓縮
}
後記
只是作為練手熟悉的一個Demo,必然是有很多問題。而且還有很多其他的配置,如:攔截器、其他的Source、Channel、Sink這裡就沒有一一的進行演示了,可參考已有的進行相應配置即可。最後還有貌似每次熟悉一個新東西都會踩很多坑啊。
下面簡單記錄幾個所遇到的坑:
java.lang.NullPointerException:
Expected timestamp in the Flume event headers, but it was null
出現這個異常是因為使用時間戳保持日誌時,沒有進行設定。新增以下這行就可以了:
// 使用時間戳配置
sinkContext.put("hdfs.useLocalTimeStamp", "true");
許可權問題:
如果日誌所在的目錄因為許可權無法讀取,這個就需要手動設定了,或者定期的複製到某個有許可權的路徑下。