如何使用Java API读写HDFS

如题所述

package com.wyc.hadoop.fs;
 
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.Date;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
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.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
 
public class FSOptr {
 
    /**
     * @param args
     */
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Configuration conf = new Configuration();
        makeDir(conf);
        rename(conf);
        delete(conf);
 
    }
 
    // åˆ›å»ºæ–‡ä»¶ç›®å½•
    private static void makeDir(Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        Path dir = new Path("/user/hadoop/data/20140318");
        boolean result = fs.mkdirs(dir);// åˆ›å»ºæ–‡ä»¶å¤¹
        System.out.println("make dir :" + result);
 
        // åˆ›å»ºæ–‡ä»¶ï¼Œå¹¶å†™å…¥å†…容
        Path dst = new Path("/user/hadoop/data/20140318/tmp");
        byte[] buff = "hello,hadoop!".getBytes();
        FSDataOutputStream outputStream = fs.create(dst);
        outputStream.write(buff, 0, buff.length);
        outputStream.close();
        FileStatus files[] = fs.listStatus(dst);
        for (FileStatus file : files) {
            System.out.println(file.getPath());
        }
        fs.close();
    }
 
    // é‡å‘½åæ–‡ä»¶
    private static void rename(Configuration conf) throws Exception {
 
        FileSystem fs = FileSystem.get(conf);
        Path oldName = new Path("/user/hadoop/data/20140318/1.txt");
        Path newName = new Path("/user/hadoop/data/20140318/2.txt");
        fs.rename(oldName, newName);
 
        FileStatus files[] = fs.listStatus(new Path(
                "/user/hadoop/data/20140318"));
        for (FileStatus file : files) {
            System.out.println(file.getPath());
        }
        fs.close();
    }
 
    // åˆ é™¤æ–‡ä»¶
    @SuppressWarnings("deprecation")
    private static void delete(Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/user/hadoop/data/20140318");
        if (fs.isDirectory(path)) {
            FileStatus files[] = fs.listStatus(path);
            for (FileStatus file : files) {
                fs.delete(file.getPath());
            }
        } else {
            fs.delete(path);
        }
 
        // æˆ–者
        fs.delete(path, true);
 
        fs.close();
    }
 
    /**
     * ä¸‹è½½,将hdfs文件下载到本地磁盘
     * 
     * @param localSrc1
     *            æœ¬åœ°çš„文件地址，即文件的路径
     * @param hdfsSrc1
     *            å­˜æ”¾åœ¨hdfs的文件地址
     */
    public boolean sendFromHdfs(String hdfsSrc1, String localSrc1) {
 
        Configuration conf = new Configuration();
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsSrc1), conf);
            Path hdfs_path = new Path(hdfsSrc1);
            Path local_path = new Path(localSrc1);
 
            fs.copyToLocalFile(hdfs_path, local_path);
 
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
 
    /**
     * ä¸Šä¼ ï¼Œå°†æœ¬åœ°æ–‡ä»¶copy到hdfs系统中
     * 
     * @param localSrc
     *            æœ¬åœ°çš„文件地址，即文件的路径
     * @param hdfsSrc
     *            å­˜æ”¾åœ¨hdfs的文件地址
     */
    public boolean sendToHdfs1(String localSrc, String hdfsSrc) {
        InputStream in;
        try {
            in = new BufferedInputStream(new FileInputStream(localSrc));
            Configuration conf = new Configuration();// å¾—到配置对象
            FileSystem fs; // æ–‡ä»¶ç³»ç»Ÿ
            try {
                fs = FileSystem.get(URI.create(hdfsSrc), conf);
                // è¾“出流，创建一个输出流
                OutputStream out = fs.create(new Path(hdfsSrc),
                        new Progressable() {
                            // é‡å†™progress方法
                            public void progress() {
                                // System.out.println("上传完一个设定缓存区大小容量的文件！");
                            }
                        });
                // è¿žæŽ¥ä¸¤ä¸ªæµï¼Œå½¢æˆé€šé“，使输入流向输出流传输数据,
                IOUtils.copyBytes(in, out, 10240, true); // in为输入流对象，out为输出流对象，4096为缓冲区大小，true为上传后关闭流
                return true;
            } catch (IOException e) {
                e.printStackTrace();
            }
 
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        return false;
    }
 
    /**
     * ç§»åŠ¨
     * 
     * @param old_st原来存放的路径
     * @param new_st移动到的路径
     */
    public boolean moveFileName(String old_st, String new_st) {
 
        try {
 
            // ä¸‹è½½åˆ°æœåŠ¡å™¨æœ¬åœ°
            boolean down_flag = sendFromHdfs(old_st, "/home/hadoop/文档/temp");
            Configuration conf = new Configuration();
            FileSystem fs = null;
 
            // åˆ é™¤æºæ–‡ä»¶
            try {
                fs = FileSystem.get(URI.create(old_st), conf);
                Path hdfs_path = new Path(old_st);
                fs.delete(hdfs_path);
            } catch (IOException e) {
                e.printStackTrace();
            }
 
            // ä»ŽæœåŠ¡å™¨æœ¬åœ°ä¼ åˆ°æ–°è·¯å¾„
            new_st = new_st + old_st.substring(old_st.lastIndexOf("/"));
            boolean uplod_flag = sendToHdfs1("/home/hadoop/文档/temp", new_st);
 
            if (down_flag && uplod_flag) {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }
 
    // copy本地文件到hdfs
    private static void CopyFromLocalFile(Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        Path src = new Path("/home/hadoop/word.txt");
        Path dst = new Path("/user/hadoop/data/");
        fs.copyFromLocalFile(src, dst);
        fs.close();
    }
 
    // èŽ·å–给定目录下的所有子目录以及子文件
    private static void getAllChildFile(Configuration conf) throws Exception {
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/user/hadoop");
        getFile(path, fs);
    }
 
    private static void getFile(Path path, FileSystem fs)throws Exception {
        FileStatus[] fileStatus = fs.listStatus(path);
        for (int i = 0; i < fileStatus.length; i++) {
            if (fileStatus[i].isDir()) {
                Path p = new Path(fileStatus[i].getPath().toString());
                getFile(p, fs);
            } else {
                System.out.println(fileStatus[i].getPath().toString());
            }
        }
    }
     
     
    //判断文件是否存在
    private static boolean isExist(Configuration conf,String path)throws Exception{
        FileSystem fileSystem = FileSystem.get(conf);
        return fileSystem.exists(new Path(path));
    }
     
    //获取hdfs集群所有主机结点数据
    private static void getAllClusterNodeInfo(Configuration conf)throws Exception{
        FileSystem fs = FileSystem.get(conf);
        DistributedFileSystem hdfs = (DistributedFileSystem)fs;
        DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
        String[] names = new String[dataNodeStats.length];
        System.out.println("list of all the nodes in HDFS cluster:"); //print info
 
        for(int i=0; i < dataNodeStats.length; i++){
            names[i] = dataNodeStats[i].getHostName();
            System.out.println(names[i]); //print info
 
        }
    }
     
    //get the locations of a file in HDFS
    private static void getFileLocation(Configuration conf)throws Exception{
        FileSystem fs = FileSystem.get(conf);
        Path f = new Path("/user/cluster/dfs.txt");
        FileStatus filestatus = fs.getFileStatus(f);
        BlockLocation[] blkLocations = fs.getFileBlockLocations(filestatus,0,filestatus.getLen());
        int blkCount = blkLocations.length;
        for(int i=0; i < blkCount; i++){
            String[] hosts = blkLocations[i].getHosts();
            //Do sth with the block hosts
 
            System.out.println(hosts);
        }
    }
     
     //get HDFS file last modification time
    private static void getModificationTime(Configuration conf)throws Exception{
        FileSystem fs = FileSystem.get(conf);
        Path f = new Path("/user/cluster/dfs.txt");
        FileStatus filestatus = fs.getFileStatus(f);
         
        long modificationTime = filestatus.getModificationTime(); // measured in milliseconds since the epoch
 
        Date d = new Date(modificationTime);
        System.out.println(d);
    }
     
}

温馨提示：内容为网友见解，仅供参考

当前网址：https://11.t2y.org/zz/8v828mpp78vms24f4p.html