hadoop 文件上传

hdsf 本质上就是一个分布式文件系统，只是相对于普通计算机来说，它可以很容易横向扩展，自带高可用机制。

我们要在Hadoop做MapReduce计算的时候，就需要把写好的程序打成jar包放到hdfs上。hadoop提供多种方式方式让你能够把文件放入hdfs，比如自带的shell命令行客户端put命令，java客户端的FileSystem，REST的HDFS API（WebHDFS与HttpFS）。

命令行上传

hadoop shell 的 put 命令直接把本地文件上传到hdfs。

使用方法：hadoop fs -put <localsrc> ... <dst>从本地文件系统中复制单个或多个源路径到目标文件系统。也支持从标准输入中读取输入写入目标文件系统。
hadoop fs -put localfile /user/hadoop/hadoopfile
hadoop fs -put localfile1 localfile2 /user/hadoop/hadoopdir
hadoop fs -put localfile hdfs://host:port/hadoop/hadoopfile
hadoop fs -put - hdfs://host:port/hadoop/hadoopfile
从标准输入中读取输入。
返回值：成功返回0，失败返回-1。

hadoop shell 官方文档：http://hadoop.apache.org/docs/r1.0.4/cn/hdfs_shell.html

Java客户端上传

我们做工程化、平台化后，需要通过程序与hadoop进行交换，hadoop提供了FileSystem 工具类供我们使用。

FileSystem官方文档：http://hadoop.apache.org/docs/current/api/org/apache/hadoop/fs/FileSystem.html

普通上传

普遍上传很简单，直接构建FileSystem的实例，调用API上传即可。需要注意的是fs.defaultFS这个配置项的值是活动的namenode地址+端口。

eg.1

public class CopyToHDFS {public static void main(String[] args) throws IOException {Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://active.namenode.host:9000");FileSystem fs = FileSystem.get(conf);fs.copyFromLocalFile(new Path("/home/lance/log.txt"), new Path("/log"));}
}

eg.2

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;import java.io.File;
import java.io.IOException;
import java.io.InputStream;public class HdfsUpload {public static void main(String[] args) throws IOException {Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://active.namenode.host:9000");FileSystem fs = FileSystem.get(conf);String remoteDst = "/user/hive/devbak/itclj/abc112234.txt";File inputFile = new File("D:/abc112233.txt");InputStream in = FileUtils.openInputStream(inputFile);FSDataOutputStream out = fs.create(new Path(remoteDst));IOUtils.copyBytes(in, out, conf);}}

eg.3 通过代理账号上传

package com.itclj;import com.itclj.kerberos.KerberosLogin;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.security.PrivilegedExceptionAction;/*** @Author lujun.chen* @Date 2020/2/1 13:33* @Version 1.0*/
public class HdfsUpload {public static void main(String[] args) throws IOException, InterruptedException {KerberosLogin login = new KerberosLogin();login.login();//直接上传//upload();//通过代理账号上传userProxyUpload();}public static void upload() throws IOException {Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://dptestservice1");//HA 模式的Hadoop集群//FileSystem fs = FileSystem.get(new URI("hdfs://s01cq-app-196-43-msxf.host:9000"), conf, "hive");FileSystem fs = FileSystem.get(conf);String remoteDst = "/user/hive/devbak/itclj/abc112234.txt";File inputFile = new File("D:/abc112233.txt");InputStream in = FileUtils.openInputStream(inputFile);FSDataOutputStream out = fs.create(new Path(remoteDst));IOUtils.copyBytes(in, out, conf);out.close();in.close();fs.close();System.out.println("===================");}public static void userProxyUpload() throws IOException, InterruptedException {String userProxy = "dsp";UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();UserGroupInformation proxyUser = UserGroupInformation.createProxyUser(userProxy, currentUser);boolean rest = proxyUser.doAs(new PrivilegedExceptionAction<Boolean>() {public Boolean run() throws Exception {upload();return true;}});System.out.print("rest=" + rest);}}

eg.4 上传限流

@Override
public HdfsOptVO upload(String path, InputStream inputStream) {try (FileSystem fs = buildFileSystem()) {try (FSDataOutputStream out = fs.create(new Path(path))) {DataTransferThrottler throttler = new DataTransferThrottler(hdfsDataTransferThrottlerSize);//限流byte buf[] = new byte[hdfsBufferSize];int bytesRead = inputStream.read(buf);while (bytesRead >= 0) {out.write(buf, 0, bytesRead);bytesRead = inputStream.read(buf);throttler.throttle(bytesRead);}//IOUtils.copyBytes(inputStream, out, 4096);inputStream.close();}} catch (Exception ex) {logger.error("HDFS文件上传失败，path: {}", path, ex);throw new DspException(CodeEnum.HDFS_OPT_UPLOAD_ERROR.getCode(), "文件上传出错，" + ex.getMessage());}return new HdfsOptVO.Builder().hdfsUri(HDFS_PROTOCOL + path).build();
}

HA模式的hadoop集群文件上传

HA模式的Hadoop集群文件上传，代码和普通模式都一样的，fs.defaultFS配置项改为集群名称(hdfs://dptestservice1)，但是单纯改这个值肯定是不行的，系统会报如下错误：

Exception in thread "main" java.lang.IllegalArgumentException: java.net.UnknownHostException: dptestservice1

主要还是工程缺少必要的Hadoop属性配置，配置属性方式一般有两种：一种是直接把hadoop的配置文件放到工程目录下，在配置文件里面设置属性；一种是如上所示，用set方法设置。

以下就是所需的关键属性，属性值得和自己集群的配置的属性值一样。

core-site.xml

<configuration><property><name>fs.defaultFS</name><value>hdfs://dptestservice1</value></property>
</configuration>

hdfs-site.xml

<configuration><property><name>dfs.nameservices</name><value>dptestservice1</value></property><property><name>dfs.ha.namenodes.dptestservice1</name><value>namenode148,namenode133</value></property><property><name>dfs.namenode.rpc-address.dptestservice1.namenode148</name><value>s01cq-app-196-39-itclj.host:9000</value></property><property><name>dfs.namenode.rpc-address.dptestservice1.namenode133</name><value>s01cq-app-196-43-itclj.host:9000</value></property><property><name>dfs.client.failover.proxy.provider.dptestservice1</name><value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value></property></configuration>

package com.itclj;import com.itclj.kerberos.KerberosLogin;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;import java.io.File;
import java.io.IOException;
import java.io.InputStream;/*** @Author lujun.chen* @Date 2020/2/1 13:33* @Version 1.0*/
public class HdfsUpload {public static void main(String[] args) throws IOException {// 集群开启了开启了kerberos 认证的，需要先进行kerberos认证。KerberosLogin login = new KerberosLogin();login.login();Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://dptestservice1");//HA 模式的Hadoop集群FileSystem fs = FileSystem.get(conf);String remoteDst = "/user/hive/devbak/itclj/abc112234.txt";File inputFile = new File("D:/abc112233.txt");InputStream in = FileUtils.openInputStream(inputFile);FSDataOutputStream out = fs.create(new Path(remoteDst));IOUtils.copyBytes(in, out, conf);}}

工程代码：https://github.com/clj198606061111/hadoop-client-demo/blob/master/src/main/java/com/itclj/HdfsUpload.java

REST的API上传

hadoop提供了2套REST的API，WebHDFS与HttpFS。

两者都是基于REST的HDFS API，使得一个集群外的host可以不用安装HADOOP和JAVA环境就可以对集群内的HADOOP进行访问，并且client不受语言的限制。
WebHDFS是HDFS内置的、默认开启的一个服务，而HttpFS是HDFS一个独立的服务，若使用需要手动安装（CDH中安装HDFS时将HttpFS勾选上即可）。
WebHDFS是HortonWorks开发的，然后捐给了Apache；而HttpFS是Cloudera开发的，也捐给了Apache。
当client请求某文件时，WebHDFS会将其重定向到该资源所在的datanode，而HttpFs相等于一个“网关”，所有的数据先传输到该httpfs server，再由该httpfs server传输到client。

WebHDFS

官方文档：https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-hdfs/WebHDFS.html#Open_and_Read_a_File

HttpFS

官方文档：http://hadoop.apache.org/docs/current/hadoop-hdfs-httpfs/index.html