hdfs常用API和putMerge功能实现

所需jar包

一、URL API操作方式

import java.io.InputStream;import java.net.URL;import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;import org.apache.hadoop.io.IOUtils;import org.junit.Test;public class HDFSUrlTest {    /**     * HDFS URL API操作方式     * 不需要读取core-site.xml和hdfs-site.xml配置文件     */    // 让JAVA程序识别HDFS的URL    static {        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());    }    // 查看文件内容    @Test    public void testRead() throws Exception {        InputStream in = null;        // 文件路径        String fileUrl = "hdfs://hadoop-master.dragon.org:9000/opt/data/test/01.data";        try {            // 获取文件输入流            in = new URL(fileUrl).openStream();            // 将文件内容读取出来，打印控制台            IOUtils.copyBytes(in, System.out, 4096, false);        } finally {            IOUtils.closeStream(in);        }    }}

二、通过FileSystem API操作HDFS

HDFS工具类

import java.io.IOException;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;public class HDFSUtils {/** *     HDFS工具类 */        public static FileSystem  getFileSystem() {        //声明FileSystem        FileSystem hdfs=null;        try {                        //获取文件配置信息            Configuration conf =new Configuration();            //获取文件系统            hdfs=FileSystem.get(conf);        } catch (IOException e) {                        e.printStackTrace();        }        return hdfs;            }    }

常用操作实现类

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.hdfs.DistributedFileSystem;import org.apache.hadoop.hdfs.protocol.DatanodeInfo;import org.apache.hadoop.io.IOUtils;import org.apache.hadoop.mapred.gethistory_jsp;import org.junit.Test;public class HDFSFsTest {    /**     *      * 通过FileSystem API操作HDFS     */    // 读取文件内容    @Test    public void testRead() throws Exception {        // 获取文件系统        FileSystem hdfs = HDFSUtils.getFileSystem();        // 文件名称        Path path = new Path("/opt/data/test/touch.data");        // 打开文件输入流        FSDataInputStream inStream = hdfs.open(path);        // 读取文件到控制台显示        IOUtils.copyBytes(inStream, System.out, 4096, false);        // 关闭流        IOUtils.closeStream(inStream);    }    // 查看目录    @Test    public void testList() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        // 文件名称        Path path = new Path("/opt/data");        FileStatus[] fileStatus = hdfs.listStatus(path);        for (FileStatus file : fileStatus) {            Path p = file.getPath();            String info = file.isDir() ? "目录" : "文件";            System.out.println(info + ":" + p);        }    }    // 创建目录    @Test    public void testDirectory() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        // 要创建的目录        Path path = new Path("/opt/data/dir");        boolean isSuccessful = hdfs.mkdirs(path);// 相当于 linux下 mkdir -p                                                    // /opt/data/dir        String info = isSuccessful ? "成功" : "失败";        System.out.println("创建目录【" + path + "】" + info);    }    // 上传文件-- put copyFromLocal    @Test    public void testPut() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        // 本地文件(目录+文件名称)        Path srcPath = new Path("c:/0125.log");        // hdfs文件上传路径        Path dstPath = new Path("/opt/data/dir/");        hdfs.copyFromLocalFile(srcPath, dstPath);    }    // 创建hdfs文件并写入内容    @Test    public void testCreate() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        Path path = new Path("/opt/data/dir/touch.data");        // 创建文件并获取输出流        FSDataOutputStream fSDataOutputStream = hdfs.create(path);        // 通过输出流写入数据        fSDataOutputStream.write("你好".getBytes());        fSDataOutputStream.writeUTF("hello hadoop!");        IOUtils.closeStream(fSDataOutputStream);    }    // 文件重命名    @Test    public void testRename() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        Path oldPath = new Path("/opt/data/dir/touch.data");        Path newPath = new Path("/opt/data/dir/rename.data");        boolean flag = hdfs.rename(oldPath, newPath);        System.out.println(flag);    }    // 删除文件    public void testDelete() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        Path path = new Path("/opt/data/dir/touch.data");        boolean flag = hdfs.deleteOnExit(path);        System.out.println(flag);    }    // 删除目录    public void testDeleteDir() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        Path path = new Path("/opt/data/dir");        boolean flag = hdfs.delete(path, true);// 如果是目录第二个参数必须为true        System.out.println(flag);    }    // 查找某个文件在hdfs集群的位置    public void testLocation() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        Path path = new Path("/opt/data/test.file");        FileStatus fileStatus = hdfs.getFileStatus(path);        BlockLocation[] blockLocations = hdfs.getFileBlockLocations(fileStatus,                0, fileStatus.getLen());        for (BlockLocation blockLocation : blockLocations) {            String[] hosts = blockLocation.getHosts();            for (String host : hosts) {                System.out.print(host + " ");            }            System.out.println();        }    }    // 获取hdfs集群上所有节点名称信息    public void testCluster() throws Exception {        FileSystem hdfs = HDFSUtils.getFileSystem();        DistributedFileSystem distributedFileSystem = (DistributedFileSystem) hdfs;        DatanodeInfo[] datanodeInfos = distributedFileSystem.getDataNodeStats();        for (DatanodeInfo datanodeInfo : datanodeInfos) {            String hostName = datanodeInfo.getHostName();            System.out.println(hostName);        }    }}

三、上传合并小文件到hdfs

实现思想：循环遍历本地文件输入流

import org.apache.hadoop.conf.Configuration;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;/** *  * 向hdfs上传复制文件的过程中，进行合并文件 *  */public class PutMerge {    /**     *      * @param localDir     *            本地要上传的文件目录     * @param hdfsFile     *            HDFS上的文件名称，包括路径     */    public static void put(String localDir, String hdfsFile) throws Exception {        // 获取配置信息        Configuration conf = new Configuration();        Path localPath = new Path(localDir);        Path hdfsPath = new Path(hdfsFile);        // 获取本地文件系统        FileSystem localFs = FileSystem.getLocal(conf);        // 获取HDFS        FileSystem hdfs = FileSystem.get(conf);        // 本地文件系统指定目录中的所有文件        FileStatus[] status = localFs.listStatus(localPath);        // 打开hdfs上文件的输出流        FSDataOutputStream fSDataOutputStream = hdfs.create(hdfsPath);        // 循环遍历本地文件        for (FileStatus fileStatus : status) {            // 获取文件            Path path = fileStatus.getPath();            System.out.println("文件为：" + path.getName());            // 打开文件输入流            FSDataInputStream fSDataInputStream = localFs.open(path);            // 进行流的读写操作            byte[] buff = new byte[1024];            int len = 0;            while ((len = fSDataInputStream.read(buff)) > 0) {                fSDataOutputStream.write(buff, 0, len);            }            fSDataInputStream.close();        }        fSDataOutputStream.close();    }        public static void main(String[] args) {        String localDir="D:/logs";        String hdfsFile="hdfs://hadoop-master.dragon.org:9000/opt/data/logs.data";        try {            put(localDir,hdfsFile);        } catch (Exception e) {                        e.printStackTrace();        }}        }