Eclipse 上传 删除 下载 分析 hdfs 上的文件

Posted 2020-10-03 末日之花的备忘录

本篇讲解如何通过Eclipse 编写代码去操作分析hdfs 上的文件。

1、在eclipse 下新建Map/Reduce Project项目。如图： 
 
项目建好后，会默认加载一系列相应的jar包。 
下面还有很多包。 
2、我们新建Java 类就可以了。代码如下：

package org.hadoop.examples;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class operateHdfs {

    public static void main(String args[]) {
        // 测试 创建新文件
        String contents = "hello world 张三\n--created by lyf\n";

        // 创建文件夹
        // creatFolder("/input/lyf2/");
        // 创建文件
        // createFile("/input/demo.txt", contents);
        // 删除文件(文件和文件夹)
        // deleteFile("/input/first.txt");
        // deleteFile("/input/lyf22");
        // 上传文件到指定位置
        // uploadFile("E:/myTest.txt", "/input");
        // 重命名(文件和文件夹--文件夹下不含文件)
        // renameFile(new Path("/input/lyf"),new Path("/input/test"));
        // 获取指定路径下的所有文件
        // getUnderDirAllFile("/input/lyf");
        // 判断文件是否存在
        // findFileIsExit(new Path("/input/lyf/first.txt"));
        // 查看某个文件在HDFS集群的位置
        // findLocationOnHadoop(new Path("/input/lyf/demo.txt"));
        // 文件内容追加
        // appendFile("/input/demo.txt", "/input/demo.txt");
        // 读取指定文件
        // readFile("/input/demo.txt");
        // 写入文件 和创建文件一样
        // writeFile("/input/demo.txt");
        // 下载文件
        // downloadFile(new Path("/input/demo.txt"),new Path("D:/demo222.txt"));

    }

    // 创建文件目录
    public static void creatFolder(String folderPath) {

        // master
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");

        Path demoDir = new Path(folderPath);
        try {
            FileSystem fs = FileSystem.get(conf);
            if (fs.mkdirs(demoDir)) {
                System.out.println("文件夹创建成功！");
            } else {
                System.out.println("文件夹创建失败！");
            }
        } catch (IOException e) {
            System.out.println("出现异常！");
        }
    }

    // 创建新文件（直接生成指定路径下的first.txt，即：/eclipse/first.txt）
    public static void createFile(String dst, String contents) {
        // master
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        System.out.println("-----------:" + conf);
        try {
            FileSystem fs = FileSystem.get(conf);

            Path dstPath = new Path(dst); // 目标路径
            // 打开一个输出流
            FSDataOutputStream outputStream = fs.create(dstPath);

            outputStream.write(contents.getBytes());
            outputStream.writeUTF(contents);

            outputStream.close();
            fs.close();
            System.out.println("文件创建成功！");
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("出现异常----文件创建失败！");
        }
    }

    // 删除指定文件
    public static void deleteFile(String file) {

        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        FileSystem fs;
        try {
            fs = FileSystem.get(conf);
            Path path = new Path(file);
            fs.delete(path);
            fs.close();
            System.out.println("文件删除成功" + conf);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("出现异常---文件删除失败" + conf);
        }

    }

    // 上传指定文件到指定路径
    public static void uploadFile(String s, String d) {

        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        FileSystem fs;
        try {
            fs = FileSystem.get(conf);
            Path src = new Path(s);
            Path dst = new Path(d);
            fs.copyFromLocalFile(src, dst);
            fs.close();
            System.out.print("文件上传成功！" + conf);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.print("异常---文件上传失败！" + conf);
        }
    }

    // 获取指定路径下的所有文件
    public static void getUnderDirAllFile(String rootPath) {
        Path targetDir = new Path(rootPath);
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "192.168.1.216:9000");

        try {
            FileSystem fs = FileSystem.get(conf);
            FileStatus[] fileStatus = fs.listStatus(targetDir);
            for (FileStatus file : fileStatus) {
                System.out.println(file.getPath() + "--" + file.getGroup() + "--" + file.getBlockSize() + "--"
                        + file.getLen() + "--" + file.getModificationTime() + "--" + file.getOwner());
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("异常----文件获取失败！");
        }
    }

    // 判断文件是否存在
    public static void findFileIsExit(Path filePath) {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        try {
            FileSystem fs = FileSystem.get(conf);
            if (fs.exists(filePath)) {
                System.out.println("文件存在！");
            } else {
                System.out.println("文件不存在！");
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("出现异常！");
        }

    }

    // 读取指定路径的文件
    public static void readFile(String path) {
        Configuration conf = new Configuration();
        //conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        try {
            FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.1.216:9000"),conf);
            FSDataInputStream ins = fs.open(new Path(path));
            // readUTF 读取的文件必须是writeUTF 写入的，否则报  EOFException
            String content = ins.readUTF();
            System.out.println("文件内容" + content);
            ins.close();
            // fs.close();

        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("读取内容失败！");
        }
    }

    // 写如文件
    public static void writeFile(String path) {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        try {
            FileSystem fs = FileSystem.get(conf);
            FSDataOutputStream ous = fs.create(new Path(path));
            ous.writeUTF("欢迎写入文件！");
            ous.close();
            System.out.println("文件写入成功！");


        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("文件写入失败！");
        }
    }

    // 追加内容到
    public static void appendFile(String filePath, String addPath) {
        Configuration conf = new Configuration();
        conf.set("fs.defaultSF", "hdfs://192.168.1.216:9000");

        conf.setBoolean("dfs.support.append", true);
        try {
            FileSystem fs = FileSystem.get(URI.create(filePath), conf);
            // 要追加的文件流，inpath为文件
            InputStream in = new BufferedInputStream(new FileInputStream(filePath));
            OutputStream out = fs.append(new Path(addPath));
            IOUtils.copyBytes(in, out, 4096, true);

            System.out.println(" 文件追加成功！" + conf);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("出现异常---文件追加失败！" + conf);
        }
    }

    // 文件重命名(指定文件改成指定名称)
    public static void renameFile(Path oldName, Path newName) {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
        try {
            FileSystem fs = FileSystem.get(conf);
            if (fs.rename(oldName, newName)) {
                System.out.println("文件重命名---成功！");
            } else {
                System.out.println("文件重命名---失败！");
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("异常----文件重命名---失败！");
        }
    }

    // 查看某个文件在HDFS集群的位置
    public static void findLocationOnHadoop(Path rootPath) {

        Configuration conf = new Configuration();
        // conf.set("fs.defultFS", "hdfs://192.168.1.216:9000");

        try {
            FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.1.216:9000"), conf);
            FileStatus fileStaus = fs.getFileStatus(rootPath);
            BlockLocation[] bloLocations = fs.getFileBlockLocations(fileStaus, 0, fileStaus.getLen());
            for (int i = 0; i < bloLocations.length; i++) {
                System.out.println("block_" + i + "_location:" + bloLocations[i].getHosts()[0]);
            }
            System.out.println("获取文件在HDFS集群位置成功！");
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("异常 ----获取文件在HDFS集群位置失败！");
        }

    }

    // 下载文件到指定位置
    public static void downloadFile(Path filePath,Path downPath){
        Configuration conf = new Configuration();

        try {
            FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.1.216:9000"),conf);
            fs.copyToLocalFile(filePath, downPath);
            System.out.println("文件下载成功！");
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("----异常--- 文件下载失败！");
        }
    }   
}

View Code

其中hdfs://192.168.1.216 是HDFS文件服务器的地址，你改成你文件服务地址就好了。

这时候是已经可以运行了，但控制台会显示如下错误：

log4j:WARN No appenders could be found for logger (org.apache.Hadoop.metrics2.lib.MutableMetricsFactory). 
log4j:WARN Please initialize the log4j system properly. 
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

解决方式：在src 文件下新建 log4j.properties 文件里面的内容如下：

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

至此文件操作就已经完成了。