数据采集之Web端上传文件到Hadoop HDFS

墨蓝 2022-09-28 11:57 127阅读 0赞

# 前言 #

最近在公司接到一个任务，是关于数据采集方面的。  
需求主要有3个：

*  通过web端上传文件到HDFS;
 *  通过日志采集的方式导入到HDFS;
 *  将数据库DB的表数据导入到HDFS。

正好最近都有在这方面做知识储备。正所谓养兵千日，用兵一时啊。学习到的东西只有应用到真实的环境中才有意义不是么。

# 环境 #

这里只做模拟环境，而不是真实的线上环境，所以也很简单，如果要使用的话还需要优化优化。

*  OS Debian 8.7
 *  [Hadoop 2.6.5][]
 *  SpringBoot 1.5.1.RELEASE

说明一下，这个系统OS最好使用Linux的，然后Hadoop也推荐使用CDH发行版的，因为在兼容性、安全性、稳定性都要好于开源的版本。比如说CDH的易于升级维护，已解决好Hadoop生态其他产品的版本兼容问题，补丁更新比开源要及时（毕竟商业公司支持）等等  
还有之所以使用SpringBoot是因为快捷，方便，不用做一大堆的配置，不管是作为演示还是生产开发都挺好的。

# 项目搭建 #

这里只是做一个很简单的演示，就是在Web页面提供一个上传按钮，使用户可以将本地文件上传至Hadoop集群平台。

## pom.xml ##

首先看下**pom**文件的依赖：

<?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.hadoop</groupId>
        <artifactId>upload</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <name>upload</name>
    
        <packaging>jar</packaging>
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.5.1.RELEASE</version>
            <relativePath/> 
        </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>
    
        </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>
            </dependency>
    
    
            <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>
    
            
            <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-archetype-plugin</artifactId>
                    <version>2.2</version>
                </plugin>
    
                <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.apache.maven.plugins</groupId>
                    <artifactId>maven-jar-plugin</artifactId>
                    <version>2.5</version>
                    <configuration>
                        <outputDirectory>${basedir}</outputDirectory>
                    </configuration>
                </plugin>
    
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    </project>

我们就是添加了一个**SpringBoot**和**Hadoop Client**的依赖。其他的是一些测试相关的。关于这个Hadoop Client它提供了一些开发Hadoop应用所需的所有依赖，可以参考之前的一篇博客：[Hadoop 2.x Maven开发环境搭建][Hadoop 2.x Maven]

# 首页 #

首页界面就只是提供一个上传表单按钮：  
index.html

<!doctype html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <meta name="viewport" content="width=device-width, user-scalable=no, initial-scale=1.0, maximum-scale=1.0, minimum-scale=1.0">
        <meta http-equiv="X-UA-Compatible" content="ie=edge">
        <title>Upload</title>
    </head>
    <body>
    <form action="/upload" method="post" enctype="multipart/form-data">
        <p>
            文件：<input type="file" name="file">
        </p>
        <p>
            <input type="submit" value="上传">
        </p>
    </form>
    </body>
    </html>

然后在**Controller**提供一个接口进行访问首页：  
HomeController.java

@Controller
    @RequestMapping(value = "/")
    public class HomeController { 
    
        public ModelAndView home() {
            return new ModelAndView("index");
        }
    
    }

# 上传 #

上传的逻辑也很简单，就是使用`SpringBoot`上传文件的形式先将文件接收到后台，然后调用`Hadoop`提供的接口API执行上传。  
上传接口UploadController.java

@Controller
    public class UploadController { 
    
        @PostMapping("/upload")
        @ResponseBody
        public String handleFileUpload(@RequestParam("file") MultipartFile file) {
    
            if (!file.isEmpty()) {
                try {
                    String originalFilename = file.getOriginalFilename();
    
                    BufferedOutputStream out = new BufferedOutputStream(
                            new FileOutputStream(
                                    new File(originalFilename)
                            )
                    );
    
                    out.write(file.getBytes());
    
                    out.flush();
                    out.close();
    
                    String destFileName = "/user/hadoop/" + originalFilename;
    
                    Upload.main(new String[]{originalFilename, destFileName});
    
                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                    return "上传失败，" + e.getMessage();
                } catch (IOException e) {
                    e.printStackTrace();
                    return "上传失败, " + e.getMessage();
                }
    
    
                return "上传成功";
    
            } else {
                return "上传失败，文件为空。";
            }
    
        }
    
    }

最后我们在提供一个类来操作**Hadoop**接口。  
Upload.java

public class Upload { 
    
        public static final String FS_DEFAULT_FS = "fs.defaultFS";
        public static final String HDFS_HOST = "hdfs://192.168.1.2:9000";
        public static final String CROSS_PLATFORM = "mapreduce.app-submission.cross-platform";
    
    
        public static void main(String[] args) throws IOException {
    
            Configuration conf = new Configuration();
    
            conf.setBoolean(CROSS_PLATFORM, true);
            conf.set(FS_DEFAULT_FS, HDFS_HOST);
    
            GenericOptionsParser optionsParser = new GenericOptionsParser(conf, args);
    
            String[] remainingArgs = optionsParser.getRemainingArgs();
            if (remainingArgs.length < 2) {
                System.err.println("Usage: upload <source> <dest>");
                System.exit(2);
            }
    
            Path source = new Path(args[0]);
            Path dest = new Path(args[1]);
    
            FileSystem fs = FileSystem.get(conf);
    
            fs.copyFromLocalFile(true, false, source, dest);
        }
    }

其中的**fs.defaultFS**属性需要与集群Master NameNode节点中配置的一直。该属性配置一般在`etc/hadoop/core-site.xml`文件中进行定义。  
可以看到我们实际的操作很简单，就只是调用**Hadoop**的FileSystem接口中的`copyFromLocalFile`方法，该方法参数说明：

*  第一个参数：表示是否删除本地的源文件，也就是上传文件后是否保留原文件，这里为了避免后续文件越来越多，就直接采用上传成功就删除的方式。
 *  第二个参数：表示是否覆盖已存在的文件，这里**false**表示不覆盖，如果HDFS集群中已存在该文件，就提示上传失败。
 *  第三个参数：源文件路径
 *  第四个参数：上传到HDFS指定的路径

# 后记 #

当然上传的方式肯定不止这一种，比如：通过Hadoop的rest接口调用PUT也可以上传，还有Python等其他语言也有相应的API接口等等

如果是要做成平台的话，这样肯定是远远不够的，每个用户都可以上传就需要做好隔离措施，我们可以采用HDFS目录隔离的方式，不过我觉得这样不够好，最好采用CDH支持的**kerberos**进行授权认证的方式比较好。开源的**Hadoop**默认只支持**Simple**的形式，也就是与操作系统一致的用户验证。

[Hadoop 2.6.5]: http://www.apache.org/dyn/closer.cgi/hadoop/common/hadoop-2.6.5/hadoop-2.6.5.tar.gz
[Hadoop 2.x Maven]: http://blog.csdn.net/lusyoe/article/details/57157677