投稿
  1. 极客之音首页

数据采集之Web端导入日志文件到HadoopHDFS

小半 • • 未分类 • 阅读 209

导读：本篇文章讲解 数据采集之Web端导入日志文件到HadoopHDFS，希望对大家有帮助，欢迎收藏，转发！站点地址：www.bmabk.com

前言

接上一篇，这一篇简单的记录一下如何在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 <lusyoe@163.com>
 * 版本：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 <lusyoe@163.com>
 * 版本：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");

权限问题： 如果日志所在的目录因为权限无法读取，这个就需要手动设置了，或者定期的复制到某个有权限的路径下。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/2031.html

赞 (0)
小半的头像小半
0 0

相关推荐

极客之音——专业性很强的中文编程技术网站，欢迎收藏到浏览器，订阅我们！