使用Java程序创建Hadoop-Client操作Hadoop

    
        
            
            
                org.apache.hadoop
                hadoop-client
                ${hadoop.version}
            
        
    

为了方便依赖版本的修改在properties中添加版本参数

    
        11
        11
        2.7.7
    

最后在Xu_Hadoop添加依赖，因为在管理中已经声明了版本因此在这里就不用再设置版本

    
        org.apache.hadoop
        hadoop-client
    

	
    public static FileSystem getfileSystem(){
        //HDFS关键类FileSystem
        //连接的URI
        URI uri = URI.create("hdfs://192.168.1.200:8020");
        //相关配置
        Configuration conf = new Configuration();
        //可以设置副本个数如:conf.set("dfs.replication","3");
        //客户端名称
        String user = "root";
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(uri, conf, user);
        } catch (IOException e) {
            logger.error("连接HDFS失败",e);
        } catch (InterruptedException e) {
            logger.error("连接HDFS失败",e);
        }
        return fileSystem;
    }

    Path src = new Path("C:\Users\xpf\Desktop\笔记\Java.txt");
    Path dst = new Path("/newFolder/");
    
    public static void upFile(FileSystem fileSystem,Path src,Path dst){
        try {
            fileSystem.copyFromLocalFile(src, dst);
        } catch (IOException e) {
            logger.error("上传文件失败");
        }
    }

    Path src = new Path("/newFolder/java.txt");
    Path dst = new Path("C:\Users\xpf\Desktop\笔记\");
    
    public static void downFile(FileSystem fileSystem,Path src,Path dst){
        try {
            fileSystem.copyToLocalFile(src, dst);
        } catch (IOException e) {
            logger.error("下载文件失败");
        }
    }

    Path src = new Path("/newFolder/python.txt");
    
    public static void createFile(FileSystem fileSystem,Path src){
        try {
            fileSystem.createNewFile(src);
        } catch (IOException e) {
            logger.error("文件创建失败");
        }
    }java

    Path src = new Path("/newFolder/");
    
    public static void deleteFile(FileSystem fileSystem,Path src){
        try {
            fileSystem.delete(src,true);
        } catch (IOException e) {java
            logger.error("文件删除失败");
        }
    }

    Path src = new Path("/newFolder/python.txt");
    Path dst = new Path("/newFolder/html.txt");
    
    public static void alterFile(FileSystem fileSystem,Path src,Path dst){
        try {
            fileSystem.rename(src,dst);
        } catch (IOException e) {java
            logger.error("修改名称失败");
        }
    }

	
    private static void printAsLS(FileStatus fileStatus) {
        StringBuilder builder = new StringBuilder();
        // 添加文件类型
        builder.append(fileStatus.isFile() ? "-" : "d");
        // 添加权限
        FsPermission permission = fileStatus.getPermission();
        builder.append(permission.toString());
        // 添加副本数
        short replication = fileStatus.getReplication();
        builder.append("t")
                .append(replication);
        //添加用户
        builder.append("t")
                .append(fileStatus.getOwner());
        //添加组
        builder.append("t")
                .append(fileStatus.getGroup());
        // 添加文件长度
        builder.append("t")
                .append(fileStatus.getLen());
        // 添加修改时间
        long time = fileStatus.getModificationTime();
        builder.append("t")
                .append(DateFormatUtils.formatYMDHM(time));
        // 添加文件路径java
        builder.append("t")
                .append(fileStatus.getPath().toString().substring(18));
        System.out.println(builder);
    }

观察源码不难发现，其实文件上传的下载都是运用了流的形式将文件转化成字节流进行传输，实现原理如下：

Path hdfsPath = new Path("/python.txt");
    String upFile = "C:\Users\xpf\Desktop\笔记\pyton\";
    String localDown = "C:\Users\xpf\Desktop\笔记\Java\";

    
    private static void get(FileSystem fileSystem, Path hdfsPath, String localDown) throws IOException {
        // HDFS  ->  输入流 -> Java程序 -> 输出流  -> 本地文件
        FSDataInputStream inputStream1 = fs.open(hdfsPath);
        FileOutputStream outputStream1 = new FileOutputStream(localDown);
        IOUtils.copyBytes(inputStream1, outputStream1, 4096);
    }
    
    
    private static void put(FileSystem fileSystem, Path hdfsPath, String upFile) throws IOException {
        //  文件  -> 输入流  -> Java程序(客户端) -> 输出流 -> HDFS
        // 1. 创建输入流读取本地文件
        FileInputStream inputStream = new FileInputStream(localFile);
        // 2. 创建输出流 用于将文件写到HDFS
        FSDataOutputStream outputStream = fs.create(hdfsPath);
        // 3. 拼接流
        IOUtils.copyBytes(inputStream, outputStream, 4096);
    }