Hadoop HDFS JAVA API 实战

怼烎@ 2022-05-16 14:25 214阅读 0赞

# JAVA API 远程调用Hadoop #

通过java Api远程调用Hadoop，实现上传文件，下载文件，删除文件等功能。

## 第一步 在pom.xml文件上添加依赖 ##

<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-common</artifactId>
    		<version>2.7.6</version>
    		<exclusions>
    			<exclusion>
    				<artifactId>slf4j-log4j12</artifactId>
    				<groupId>org.slf4j</groupId>
    			</exclusion>
    		</exclusions>
    		
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-archives</artifactId>
    		<version>2.7.6</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-auth</artifactId>
    		<version>2.7.6</version>
    		<exclusions>
    			<exclusion>
    				<artifactId>slf4j-log4j12</artifactId>
    				<groupId>org.slf4j</groupId>
    			</exclusion>
    		</exclusions>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-datajoin</artifactId>
    		<version>2.7.6</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-distcp</artifactId>
    		<version>2.7.6</version>
    	</dependency>
    
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-hdfs</artifactId>
    		<version>2.7.6</version>
    		<exclusions>
    			<exclusion>
    				<artifactId>netty-all</artifactId>
    				<groupId>io.netty</groupId>
    			</exclusion>
    		</exclusions>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-annotations</artifactId>
    		<version>2.7.6</version>
    		<exclusions>
    			<exclusion>
    				<groupId>jdk.tools</groupId>
    				<artifactId>jdk.tools</artifactId>
    			</exclusion>
    		</exclusions>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-client</artifactId>
    		<version>2.7.6</version>
    		<exclusions>
    			<exclusion>
    				<artifactId>slf4j-log4j12</artifactId>
    				<groupId>org.slf4j</groupId>
    			</exclusion>
    		</exclusions>
    	</dependency>

## 第二步 创建测试类FileSystemCat ##

package com.example.wfhadoop.hdfs;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FileStatus;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IOUtils;
    
    import java.io.*;
    import java.net.URI;
    import java.net.URISyntaxException;
    
    /**
     * @author wufei
     * @create 2018-10-30 11:05
     **/
    public class FileSystemCat {
    
        static Configuration conf = new Configuration();
        static FileSystem hdfs;
        static {
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://hdn1:9000");
            try {
                try {
                    hdfs = FileSystem.get(URI.create("hdfs://hdn1:9000"),conf,"hadoop");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        public static void main(String[] args) {
            try {
                String dir = "/Hadoop/wftest";
                String src = "E:\\hadoofile\\ok1.txt";
                //显示文件目录
                //listFiles("hdfs://hdn1:9000/Hadoop/Input/");
                //创建文件目录
                  createDir(dir);
                //上传文件
                //uploadFile(src,dir);
                //下载文件
                //down(src,"hdfs://hdn1:9000/Hadoop/Input/ok1.txt");
                //删除文件
                //deleteFile("/Hadoop/Input/ok.txt");
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    
    
    
        /**
         * 下载文件
         * @param localSrc  E:\hadoofile\ok1.txt
         * @param hdfsDst  hdfs://hdn1:9000/Hadoop/Input/ok1.txt
         */
        public static void down(String localSrc, String hdfsDst){
            FSDataInputStream   in = null;
            try {
                in = hdfs.open(new Path(hdfsDst));
                OutputStream out = new FileOutputStream(localSrc);
                IOUtils.copyBytes(in, out, 4096, true);
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if(in != null){
                    IOUtils.closeStream(in);
                }
            }
        }
    
        public static void createDir(String dir) throws Exception {
            Path path = new Path(dir);
            if (hdfs.exists(path)) {
                return;
            }
            hdfs.mkdirs(path);   
             }
            /**
             * 文件列表
             * @param dirName 文件目录
             * @throws IOException
             */
        public static void listFiles(String dirName) throws IOException {
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://hdn1:9000");
            FileSystem hdfs = null;
            FSDataInputStream in = null;
            hdfs = FileSystem.get(conf);
            Path path = new Path(dirName);
            FileStatus[] status = hdfs.listStatus(path);
            System.out.println(dirName + " has all files:");
            for (int i = 0; i < status.length; i++) {
                System.out.println(status[i].getPath().toString());
            }
        }
    
        /**
         * 文件上传
         * @param localSrc
         * @param hdfsDst
         * @throws IOException
         */
        public static void uploadFile(String localSrc, String hdfsDst) throws IOException {
            Path src = new Path(localSrc);
            Path dst = new Path(hdfsDst);
            if (!(new File(localSrc)).exists()) {
                return;
            }
            if (!hdfs.exists(dst)) {
                return;
            }
            String dstPath = dst.toUri() + "/" + src.getName();
            if (hdfs.exists(new Path(dstPath))) {
            }
            hdfs.copyFromLocalFile(src, dst);
            FileStatus files [] = hdfs.listStatus(dst);
            for (FileStatus file : files) {
                System.out.println(file.getPath());
            }
        }
    
        /**
         * 删除文件
         * @param fileName
         * @throws IOException
         */
        public static void deleteFile(String fileName) throws IOException {
            Path f = new Path(fileName);
            boolean isExists = hdfs.exists(f);
            if (isExists) { // if exists, delete
                boolean isDel = hdfs.delete(f, true);
            } else {
                System.out.println(fileName + "  exist? \t" + isExists);
            }
        }
    }

## 第三步 测试 ##

1.执行上传文件
       查看文件是否上传成功
      [hadoop@hdn1 ~]$ hadoop fs -ls /Hadoop/Input
    	Found 2 items
    	-rw-r--r--   3 hadoop supergroup         17 2018-10-31 09:39 /Hadoop/Input/ok1.txt
    	-rw-r--r--   2 hadoop supergroup         34 2018-10-30 09:39 /Hadoop/Input/wordcount.txt