1、HDFS客户端环境准备

根据自己电脑的操作系统拷贝对应的编译后的hadoop jar包到非中文路径，例如：D:\Develop\hadoop-2.7.2。

配置HADOOP_HOME环境变量。

配置Path环境变量。

导入相应的依赖坐标+日志添加。


   	
    		
     
      org.apache.logging.log4j
     		
     
      log4j-core
     		
     
      2.8.2
     	
    	
    		
     
      org.apache.hadoop
     		
     
      hadoop-common
     		
     
      2.7.2
     	
    	
    		
     
      org.apache.hadoop
     		
     
      hadoop-client
     		
     
      2.7.2
     	
    	
    		
     
      org.apache.hadoop
     		
     
      hadoop-hdfs
     		
     
      2.7.2

在项目的src/main/resources目录下，新建一个文件，命名为log4j.properties，内容如下：

log4j.rootLogger=INFO, stdoutlog4j.appender.stdout=org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.layout=org.apache.log4j.PatternLayoutlog4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%nlog4j.appender.logfile=org.apache.log4j.FileAppenderlog4j.appender.logfile.File=target/spring.loglog4j.appender.logfile.layout=org.apache.log4j.PatternLayoutlog4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

2、HDFS的API操作

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.*;import org.apache.hadoop.io.IOUtils;import org.junit.After;import org.junit.Before;import org.junit.Test;import java.io.File;import java.io.FileInputStream;import java.io.FileOutputStream;public class HDFSClient {
       FileSystem fs = null;    Configuration conf = null;    // 客户端去操作HDFS时，是有一个用户身份的。    // 默认情况下，HDFS客户端API会从JVM中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME=root，root为用户名称。    @Before    public void before() throws Exception {
           conf = new Configuration();        conf.set("fs.defaultFS", "hdfs://yangqian:9000");        // 设置操作用户        System.setProperty("HADOOP_USER_NAME", "yangqian");        fs = FileSystem.get(conf);        // 同时指定文件系统地址和操作用户        // fs = FileSystem.get(new URI("hdfs://yangqian:9000"), conf, "yangqian");    }    @After    public void after() throws Exception {
           if (fs != null) {
               fs.close();        }    }    // 创建目录    @Test    public void mkdir() throws Exception {
           System.out.println(fs.mkdirs(new Path("/hdfs")));    }    // 删除目录或文件    @Test    public void delete() throws Exception {
           System.out.println(fs.delete(new Path("/hdfs"), true));    }    // 文件的上传和下载    @Test    public void getAndPut() throws Exception {
           // 上传        fs.copyFromLocalFile(new Path("/home/yangqian/a.txt"), new Path("/hdfs/a.txt"));        // 下载        fs.copyToLocalFile(new Path("/hdfs/a.txt"), new Path("/home/yangqian/a.txt"));    }    // 文件名更改    @Test    public void rename() throws Exception{
           fs.rename(new Path("/hdfs/url.csv"), new Path("/hdfs/url.txt"));    }    // 文件详情查看    @Test    public void listFiles() throws Exception {
           RemoteIterator
   
     listFiles = fs.listFiles(new Path("/"), true);        while (listFiles.hasNext()){
               LocatedFileStatus status = listFiles.next();            // 输出详情            // 文件名称            System.out.println(status.getPath().getName());            // 长度            System.out.println(status.getLen());            // 权限            System.out.println(status.getPermission());            // 分组            System.out.println(status.getGroup());            // 获取存储的块信息            BlockLocation[] blockLocations = status.getBlockLocations();            for (BlockLocation blockLocation : blockLocations) {
                   // 获取块存储的主机节点                String[] hosts = blockLocation.getHosts();                for (String host : hosts) {
                       System.out.println(host);                }            }            System.out.println("-----------分割线----------");        }    }    // 文件和文件夹判断    @Test    public void listStatus() throws Exception{
           // 2 判断是文件还是文件夹        FileStatus[] listStatus = fs.listStatus(new Path("/hdfs/url.txt"));        for (FileStatus fileStatus : listStatus) {
               // 如果是文件            if (fileStatus.isFile()) {
                   System.out.println("f:"+fileStatus.getPath().getName());            }else {
                   System.out.println("d:"+fileStatus.getPath().getName());            }        }    }    // 使用流上传文件    @Test    public void putFileToHDFS() throws Exception {
           // 创建输入流        FileInputStream fis = new FileInputStream(new File("/home/yangqian/url.csv"));        // 获取输出流        FSDataOutputStream fos = fs.create(new Path("/hdfs/test.txt"));        // 流对拷        IOUtils.copyBytes(fis, fos, conf);        // 关闭资源        IOUtils.closeStream(fos);        IOUtils.closeStream(fis);    }    // 使用流下载文件    @Test    public void getFileFromHDFS() throws Exception {
           // 获取输入流        FSDataInputStream fis = fs.open(new Path("/hdfs/test.txt"));        // 获取输出流        FileOutputStream fos = new FileOutputStream(new File("/home/yangqian/url123.csv"));        // 流的对拷        IOUtils.copyBytes(fis, fos, conf);        // 5 关闭资源        IOUtils.closeStream(fos);        IOUtils.closeStream(fis);    }}