HDFS入门小demo

冷不防 2023-02-17 03:21 17阅读 0赞

package cn.test.hdfs_api_demo;
    
    import java.io.File;
    import java.io.FileOutputStream;
    import java.io.IOException;
    import java.io.InputStream;
    import java.net.URI;
    import java.net.URISyntaxException;
    import java.net.URL;
    
    import org.apache.commons.io.IOUtils;
    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.FsUrlStreamHandlerFactory;
    import org.apache.hadoop.fs.LocalFileSystem;
    import org.apache.hadoop.fs.LocatedFileStatus;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.fs.RemoteIterator;
    import org.junit.Test;
    
    public class HdfsAPIDemo {
    
    	@Test
    	public void demo1() throws Exception {
    		// 第一步：注册hdfs 的url
    		URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
    		// 获取文件输入流
    		InputStream inputStream = new URL("hdfs://node01:8020/core-site.xml").openStream();
    		// 获取文件输出流
    		FileOutputStream outputStream = new FileOutputStream(new File("e:\\hdfstest\\hello.txt"));
    		// 实现文件的拷贝
    		IOUtils.copy(inputStream, outputStream);
    		// 关闭流
    		IOUtils.closeQuietly(inputStream);
    		IOUtils.closeQuietly(outputStream);
    	}
    
    	/**
    	 * 获取 FileSyttem :方式1
    	 * 
    	 * @throws Exception
    	 */
    	@Test
    	public void getFileSystem1() throws Exception {
    		// 创建Configuration对象
    		Configuration conf = new Configuration();
    		// 设置文件系统类型
    		conf.set("fs.defaultFS", "hdfs://node01:8020/");
    		// 获取指定的文件系统
    		FileSystem fileSystem = FileSystem.get(conf);
    		// 输出
    		System.out.println(fileSystem.toString());
    	}
    
    	/**
    	 * 获取 FileSyttem :方式2
    	 * 
    	 * @throws Exception
    	 */
    	@Test
    	public void getFileSystem2() throws Exception {
    
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    		System.out.println("fileSystem:" + fileSystem);
    	}
    
    	/**
    	 * 获取 FileSyttem :方式3
    	 * 
    	 * @throws Exception
    	 */
    	@Test
    	public void getFileSystem3() throws Exception {
    
    		// 创建Configuration对象
    		Configuration conf = new Configuration();
    		// 设置文件系统类型
    		conf.set("fs.defaultFS", "hdfs://node01:8020/");// 不加hdfs://,默认是file://,本地文件
    		// 获取指定的文件系统
    		FileSystem fileSystem = FileSystem.newInstance(conf);
    		// 输出
    		System.out.println(fileSystem.toString());
    	}
    
    	/**
    	 * 获取 FileSyttem :方式4
    	 * 
    	 * @throws Exception
    	 */
    	@Test
    	public void getFileSystem4() throws Exception {
    
    		FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020"), new Configuration());
    		// 输出
    		System.out.println(fileSystem.toString());
    	}
    
    	/**
    	 * 遍历
    	 */
    	@Test
    	public void listMyFiles() throws Exception {
    		// 获取FileSystem
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    
    		// 调用listFiles,获取/目录下所有的文件信息
    		RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listFiles(new Path("/"), true);// true表示递归获取
    
    		// 遍历迭代器
    		while (listFiles.hasNext()) {
    			LocatedFileStatus locatedFileStatus = listFiles.next();
    			// 获取文件的绝对路径
    			Path path = locatedFileStatus.getPath();
    			// 文件名
    			String name = path.getName();
    			// block信息 block数组
    			BlockLocation[] blockLocations = locatedFileStatus.getBlockLocations();
    			System.out.println(path + "==" + name + "block数" + blockLocations.length);
    		}
    	}
    
    	/**
    	 * 创建文件夹 @throws Exception @throws
    	 */
    	@Test
    	public void mkdirsTest() throws Exception {
    		// 获取FileSystem对象
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    		// 创建文件夹
    		boolean flag = fileSystem.mkdirs(new Path("/aaa/bbb/ccc"));
    		System.out.println("创建文件夹:" + flag);
    		// 关闭fileSystem
    		fileSystem.close();
    	}
    
    	/**
    	 * 创建文件 @throws Exception @throws
    	 */
    	@Test
    	public void mkFileTest() throws Exception {
    		// 获取FileSystem对象
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    		// 创建文件夹
    		fileSystem.create(new Path("/aaa/bbb/ccc/a.txt"));
    		// System.out.println("创建文件夹:"+flag);
    		// 关闭fileSystem
    		fileSystem.close();
    	}
    
    	/**
    	 * 文件下载
    	 * 
    	 * @throws URISyntaxException
    	 * @throws IOException
    	 */
    	@Test
    	public void downloadFile() throws IOException, URISyntaxException {
    		// 1.获取fileSystem
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    		// 2.获取hdfs的输入流
    		FSDataInputStream fis = fileSystem.open(new Path("/core-site.xml"));
    		// 3.获取本地路径输出流
    		FileOutputStream fos = new FileOutputStream(new File("e:\\hdfstest\\core-site.xml"));
    		// 4.文件的拷贝
    		IOUtils.copy(fis, fos);
    
    		// 5.关流
    		IOUtils.closeQuietly(fis);
    		IOUtils.closeQuietly(fos);
    		fileSystem.close();
    	}
    
    	/**
    	 * 文件下载2
    	 * 
    	 * @throws URISyntaxException
    	 * @throws IOException
    	 */
    	@Test
    	public void downloadFile2() throws IOException, URISyntaxException {
    		// 1.获取fileSystem
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    		// 2.调用方法实现文件的下载
    		fileSystem.copyToLocalFile(new Path("/zhangsan.txt"), new Path("e:\\hdfstest\\zhangsan3.txt"));
    
    		// 3.关流
    
    		fileSystem.close();
    	}
    
    	/**
    	 * 文件上传
    	 * 
    	 * @throws URISyntaxException
    	 * @throws IOException
    	 */
    
    	@Test
    	public void uploadFile() throws IOException, URISyntaxException {
    		// 1.获取fileSystem
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
    		// 2.调用方法实现上传
    		fileSystem.copyFromLocalFile(new Path("e:\\hdfstest\\zhangsan.txt"), new Path("/"));
    		// 关闭fileSystem
    		fileSystem.close();
    	}
    
    	/**
    	 * 文件下载 :现用户无权限,通过其他用户下载
    	 * 
    	 * @throws InterruptedException
    	 */
    
    	@Test
    	public void downloadFile3() throws IOException, URISyntaxException, InterruptedException {
    		// 1.获取fileSystem
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration(), "root");
    		// 2.调用方法实现文件的下载
    		fileSystem.copyToLocalFile(new Path("/core-site.xml"), new Path("e:\\hdfstest\\core-site3.xml"));
    
    		// 3.关流
    
    		fileSystem.close();
    	}
    
    	/**
    	 * 将本地小文件合并成大文件上传
    	 * 
    	 * @throws URISyntaxException
    	 * @throws IOException
    	 * @throws InterruptedException
    	 */
    	@Test
    	public void mergeFile() throws IOException, URISyntaxException, InterruptedException {
    		// 1.获取fileSystem
    		FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration(), "root");
    
    		// 2.获取输出流
    		FSDataOutputStream fos = fileSystem.create(new Path("/big_file.txt"));
    
    		// 3.获取本地文件系统
    		LocalFileSystem local = fileSystem.getLocal(new Configuration());
    		// 4.获取本地文件夹下所有文件详情
    		FileStatus[] fileStatuses = local.listStatus(new Path("E:\\hdfsinput"));
    		// 5.遍历每个文件,获取每个文件的输入流
    		for (FileStatus fileStatus : fileStatuses) {
    
    			FSDataInputStream fis = local.open(fileStatus.getPath());
    			// 6.将小文件数据复制到大文件
    			IOUtils.copy(fis, fos);
    			IOUtils.closeQuietly(fis);
    		}
    
    		// 7.关流
    
    		IOUtils.closeQuietly(fos);
    		local.close();
    		fileSystem.close();
    
    	}
    }