HDFS API操作

x33g5p2x  于2020-09-30 发布在 HDFS  
字(10.5k)|赞(0)|评价(0)|浏览(742)

1.maven准备

  1. <dependencies>
  2. <dependency>
  3.             <groupId>org.apache.hadoop</groupId>
  4.             <artifactId>hadoop-common</artifactId>
  5.             <version>2.8.4</version>
  6.         </dependency>
  8.         <dependency>
  9.             <groupId>org.apache.hadoop</groupId>
  10.             <artifactId>hadoop-common</artifactId>
  11.             <version>2.8.4</version>
  12.         </dependency>
  13.         <dependency>
  14.             <groupId>org.apache.hadoop</groupId>
  15.             <artifactId>hadoop-hdfs</artifactId>
  16.             <version>2.8.4</version>
  17.         </dependency>
  18.         <dependency>
  19.             <groupId>org.apache.hadoop</groupId>
  20.             <artifactId>hadoop-client</artifactId>
  21.             <version>2.8.4</version>
  22.         </dependency>
  24.         <dependency>
  25.             <groupId>org.projectlombok</groupId>
  26.             <artifactId>lombok</artifactId>
  27.             <version>1.16.10</version>
  28.         </dependency>
  30.         <dependency>
  31.             <groupId>log4j</groupId>
  32.             <artifactId>log4j</artifactId>
  33.             <version>1.2.17</version>
  34.         </dependency>
  35.         <dependency>
  36.             <groupId>org.slf4j</groupId>
  37.             <artifactId>slf4j-api</artifactId>
  38.             <version>1.7.7</version>
  39.         </dependency>
  41.         <!-- https://mvnrepository.com/artifact/junit/junit -->
  42.         <dependency>
  43.             <groupId>junit</groupId>
  44.             <artifactId>junit</artifactId>
  45.             <version>4.12</version>
  46.             <scope>test</scope>
  47.         </dependency>
  48.    </dependencies>

2.本地配置Hadoop环境

在windows系统需要配置hadoop运行环境，否则直接运行代码会出现以下问题:

  1. 解压hadoop2.8.4文件至需要安装的全英文目录下

  2. 配置hadoop的环境变量 HADOOP_HOME

    将以下配置到path变量中

  1.    D:\hadoop-2.8.4\hadoop-2.8.4\sbin
  2.    D:\hadoop-2.8.4\hadoop-2.8.4\bin

注：按照自己的安装路径进行配置

3.获取文件系统

  1. /**
  2.      * 打印本地hadoop地址值
  3.      * IO的方式写代码
  4.      */
  5.     @Test
  6.     public void intiHDFS() throws IOException {
  7.         //F2 可以快速的定位错误
  8.         // alt + enter自动找错误
  9.         //1.创建配信信息对象 ctrl + alt + v  后推前  ctrl + shitl + enter 补全
  10.         Configuration conf = new Configuration();
  12.         //2.获取文件系统
  13.         FileSystem fs = FileSystem.get(conf);
  15.         //3.打印文件系统
  16.         System.out.println(fs.toString());
  17.     }

4.遍历HDFS所有文件

  1.  //遍历HDFS所有文件
  2.     @Test
  3.     public void listFiles() throws Exception{
  4.         //1.获取filesystem对象
  5.         FileSystem fileSystem = FileSystem.get(new URI("hdfs://bigdata111:9000"), new Configuration());
  6.         //2.获取RemoteIterator 得到所有的文件或者文件夹，第一个参数指定遍历的路径，第二个can阿叔表示是否要递归遍历
  7.         RemoteIterator<LocatedFileStatus> lists = fileSystem.listFiles(new Path("/"), true);
  8.         //3.循环
  9.         while (lists.hasNext()){
  10.             LocatedFileStatus next = lists.next();
  11.             System.out.println(next.getPath().toString());
  13.             //文件的block信息
  14.             BlockLocation[] blockLocations = next.getBlockLocations();
  15.             System.out.println("block数： "+blockLocations.length);
  16.         }
  17.         fileSystem.close();
  18.     }

5.HDFS上创建文件夹

  1. /**
  2.      * hadoop fs -mkdir /xinshou
  3.      */
  4.     @Test
  5.     public void mkmdirHDFS() throws Exception {
  6.         //1.创新配置信息对象
  7.         Configuration configuration = new Configuration();
  9.         //2.链接文件系统
  10.         //final URI uri  地址
  11.         //final Configuration conf  配置
  12.         //String user   Linux用户
  13.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), configuration, "root");
  15.         //3.创建目录
  16.         fs.mkdirs(new Path("hdfs://bigdata111:9000/Good/Goog/Study"));
  18.         //4.关闭
  19.         fs.close();
  20.         System.out.println("创建文件夹成功");
  21.     }

6. 文件下载

  1.   /**
  2.      * hadoop fs -get /HDFS文件系统
  3.      * @throws Exception
  4.      */
  5.     @Test
  6.     public void getFileFromHDFS() throws Exception {
  7.         //1.创建配置信息对象  Configuration:配置
  8.         Configuration conf = new Configuration();
  10.         //2.找到文件系统
  11.         //final URI uri     ：HDFS地址
  12.         //final Configuration conf：配置信息
  13.         // String user ：Linux用户名
  14.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  16.         //3.下载文件
  17.         //boolean delSrc:是否将原文件删除
  18.         //Path src ：要下载的路径
  19.         //Path dst ：要下载到哪
  20.         //boolean useRawLocalFileSystem ：是否校验文件
  21.         fs.copyToLocalFile(false,new Path("hdfs://bigdata111:9000/README.txt"),
  22.                 new Path("F:\\date\\README.txt"),true);
  24.         //4.关闭fs
  25.         //alt + enter 找错误
  26.         //ctrl + alt + o  可以快速的去除没有用的导包
  27.         fs.close();
  28.         System.out.println("下载成功");
  29.     }

7.文件上传

  2.  /**
  3.      * 上传代码
  4.      * 注意：如果上传的内容大于128MB,则是2块
  5.      */
  6.     @Test
  7.     public void putFileToHDFS() throws Exception {
  8.         //注：import org.apache.hadoop.conf.Configuration;
  9.         //ctrl + alt + v 推动出对象
  10.         //1.创建配置信息对象
  11.         Configuration conf = new Configuration();
  13.         //2.设置部分参数
  14.         conf.set("dfs.replication","2");
  16.         //3.找到HDFS的地址
  17.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  19.         //4.上传本地Windows文件的路径
  20.         Path src = new Path("D:\\hadoop-2.7.2.rar");
  21.          //5.要上传到HDFS的路径
  22.         Path dst = new Path("hdfs://bigdata111:9000/Andy");
  24.         //6.以拷贝的方式上传，从src -> dst
  25.         fs.copyFromLocalFile(src,dst);
  27.         //7.关闭
  28.         fs.close();
  30.         System.out.println("上传成功");
  31. }

8.小文件合并至HDFS

  1. //小文件合并至HDFS
  2.     @Test
  3.     public void mergeFile() throws Exception{
  4.         //1.获取FileSystem系统
  5.         FileSystem fileSystem = FileSystem.get(new URI("hdfs://bigdata111:9000"),new Configuration());
  6.         //2.获取hdfs大文件输出流
  7.         FSDataOutputStream OutputStream = fileSystem.create(new Path("/hello.txt"));
  8.         //3.获取一个本地文件系统
  9.         LocalFileSystem local = FileSystem.getLocal(new Configuration());
  10.         //4.获取本地文件夹下所有文件的详情
  11.         FileStatus[] fileStatuses = local.listStatus(new Path("G:\\file"));
  12.         //5.遍历每个文件，获取每个文件的输入流
  13.         for (FileStatus fileStatus : fileStatuses) {
  14.             FSDataInputStream open = local.open(fileStatus.getPath());
  15.             //6.将小文件复制到大文件
  16.             IOUtils.copy(open,OutputStream);
  17.             //7.关闭流
  18.             IOUtils.closeQuietly(open);
  19.         }
  20.         IOUtils.closeQuietly(OutputStream);
  21.         local.close();
  22.         fileSystem.close();
  23.     }

9.HDFS文件夹删除

  1.  /**
  2.      * hadoop fs -rm -r /文件
  3.      */
  4.     @Test
  5.     public void deleteHDFS() throws Exception {
  6.         //1.创建配置对象
  7.         Configuration conf = new Configuration();
  9.         //2.链接文件系统
  10.         //final URI uri, final Configuration conf, String user
  11.         //final URI uri  地址
  12.         //final Configuration conf  配置
  13.         //String user   Linux用户
  14.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  16.         //3.删除文件
  17.         //Path var1   : HDFS地址
  18.         //boolean var2 : 是否递归删除
  19.         fs.delete(new Path("hdfs://bigdata111:9000/a"),false);
  21.         //4.关闭
  22.         fs.close();
  23.         System.out.println("删除成功啦");
  24.     }

10.HDFS文件名更改

  1. 	@Test
  2. 	public void renameAtHDFS() throws Exception{
  3. 		// 1 创建配置信息对象
  4. 		Configuration configuration = new Configuration();
  5. 		
  6. 		FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"),configuration, "itstar");
  7. 		
  8. 		//2 重命名文件或文件夹
  9. 		fs.rename(new Path("hdfs://bigdata111:9000/user/itstar/hello.txt"), new Path("hdfs://bigdata111:9000/user/itstar/hellonihao.txt"));
  10. 	fs.close();
  11. 	}

11.HDFS文件详情查看

  1.   /**
  2.      * 查看【文件】名称、权限等
  3.      */
  4.     @Test
  5.     public void readListFiles() throws Exception {
  6.         //1.创建配置对象
  7.         Configuration conf = new Configuration();
  9.         //2.链接文件系统
  10.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  12.         //3.迭代器
  13.         RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
  15.         //4.遍历迭代器
  16.         while (listFiles.hasNext()){
  17.             //一个一个出
  18.             LocatedFileStatus fileStatus = listFiles.next();
  20.             //名字
  21.             System.out.println("文件名：" + fileStatus.getPath().getName());
  22.             //块大小
  23.             System.out.println("大小：" + fileStatus.getBlockSize());
  24.             //权限
  25.             System.out.println("权限：" + fileStatus.getPermission());
  26.             System.out.println(fileStatus.getLen());
  29.             BlockLocation[] locations = fileStatus.getBlockLocations();
  31.             for (BlockLocation bl:locations){
  32.                 System.out.println("block-offset:" + bl.getOffset());
  33.                 String[] hosts = bl.getHosts();
  34.                 for (String host:hosts){
  35.                     System.out.println(host);
  36.                 }
  37.             }
  39.             System.out.println("------------------华丽的分割线----------------");
  40.         }

12.HDFS文件和文件夹判断

  1.    /**
  2.      * 判断是否是个文件还是目录，然后打印
  3.      * @throws Exception
  4.      */
  5.     @Test
  6.     public void judge() throws Exception {
  7.         //1.创建配置文件信息
  8.         Configuration conf = new Configuration();
  10.         //2.获取文件系统
  11.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  13.         //3.遍历所有的文件
  14.         FileStatus[] liststatus = fs.listStatus(new Path("/Andy"));
  15.         for(FileStatus status :liststatus)
  16.         {
  17.             //判断是否是文件
  18.             if (status.isFile()){
  19.                 //ctrl + d:复制一行
  20.                 //ctrl + x 是剪切一行，可以用来当作是删除一行
  21.                 System.out.println("文件:" + status.getPath().getName());
  22.             } else {
  23.                 System.out.println("目录:" + status.getPath().getName());
  24.             }
  25.         }
  26.     }

13.通过IO流操作HDFS

13.1HDFS文件上传

  1.  /**
  2.      * IO流方式上传
  3.      *
  4.      * @throws URISyntaxException
  5.      * @throws FileNotFoundException
  6.      * @throws InterruptedException
  7.      */
  8.     @Test
  9.     public void putFileToHDFSIO() throws URISyntaxException, IOException, InterruptedException {
  10.         //1.创建配置文件信息
  11.         Configuration conf = new Configuration();
  13.         //2.获取文件系统
  14.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  16.         //3.创建输入流
  17.         FileInputStream fis = new FileInputStream(new File("F:\\date\\Sogou.txt"));
  19.         //4.输出路径
  20.         //注意：不能/Andy  记得后边写个名 比如：/Andy/Sogou.txt
  21.         Path writePath = new Path("hdfs://bigdata111:9000/Andy/Sogou.txt");
  22.         FSDataOutputStream fos = fs.create(writePath);
  24.         //5.流对接
  25.         //InputStream in    输入
  26.         //OutputStream out  输出
  27.         //int buffSize      缓冲区
  28.         //boolean close     是否关闭流
  29.         try {
  30.             IOUtils.copyBytes(fis,fos,4 * 1024,false);
  31.         } catch (IOException e) {
  32.             e.printStackTrace();
  33.         }finally {
  34.             IOUtils.closeStream(fos);
  35.             IOUtils.closeStream(fis);
  36.             System.out.println("上传成功啦");
  37.         }
  38.     }

13.2 HDFS文件下载

  1. /**
  2.      * IO读取HDFS到控制台
  3.      *
  4.      * @throws URISyntaxException
  5.      * @throws IOException
  6.      * @throws InterruptedException
  7.      */
  8.     @Test
  9.     public void getFileToHDFSIO() throws URISyntaxException, IOException, InterruptedException {
  10.         //1.创建配置文件信息
  11.         Configuration conf = new Configuration();
  13.         //2.获取文件系统
  14.         FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  16.         //3.读取路径
  17.         Path readPath = new Path("hdfs://bigdata111:9000/Andy/Sogou.txt");
  19.         //4.输入
  20.         FSDataInputStream fis = fs.open(readPath);
  22.         //5.输出到控制台
  23.         //InputStream in    输入
  24.         //OutputStream out  输出
  25.         //int buffSize      缓冲区
  26.         //boolean close     是否关闭流
  27.         IOUtils.copyBytes(fis,System.out,4 * 1024 ,true);
  28.     }

13.3 定位文件读取

  1. 下载第一块
  1.    /**
  2.         * IO读取第一块的内容
  3.         *
  4.         * @throws Exception
  5.         */
  6.        @Test
  7.        public void  readFlieSeek1() throws Exception {
  8.            //1.创建配置文件信息
  9.            Configuration conf = new Configuration();
  10.    
  11.            //2.获取文件系统
  12.            FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  13.    
  14.            //3.输入
  15.            Path path = new Path("hdfs://bigdata111:9000/Andy/hadoop-2.7.2.rar");
  16.            FSDataInputStream fis = fs.open(path);
  17.    
  18.            //4.输出
  19.            FileOutputStream fos = new FileOutputStream("F:\\date\\readFileSeek\\A1");
  20.    
  21.            //5.流对接
  22.            byte[] buf = new byte[1024];
  23.            for (int i = 0; i < 128 * 1024; i++) {
  24.                fis.read(buf);
  25.                fos.write(buf);
  26.            }
  27.    
  28.            //6.关闭流
  29.            IOUtils.closeStream(fos);
  30.            IOUtils.closeStream(fis);
  31.        }
  1. 下载第二块
  1.    /**
  2.         * IO读取第二块的内容
  3.         *
  4.         * @throws Exception
  5.         */
  6.        @Test
  7.        public void readFlieSeek2() throws Exception {
  8.            //1.创建配置文件信息
  9.            Configuration conf = new Configuration();
  10.    
  11.            //2.获取文件系统
  12.            FileSystem fs = FileSystem.get(new URI("hdfs://bigdata111:9000"), conf, "root");
  13.    
  14.            //3.输入
  15.            Path path = new Path("hdfs://bigdata111:9000/Andy/hadoop-2.7.2.rar");
  16.            FSDataInputStream fis = fs.open(path);
  17.    
  18.            //4.输出
  19.            FileOutputStream fos = new FileOutputStream("F:\\date\\readFileSeek\\A2");
  20.    
  21.            //5.定位偏移量/offset/游标/读取进度 (目的：找到第一块的尾巴，第二块的开头)
  22.            fis.seek(128 * 1024 * 1024);
  23.    
  24.            //6.流对接
  25.            IOUtils.copyBytes(fis, fos, 1024);
  26.    
  27.            //7.关闭流
  28.            IOUtils.closeStream(fos);
  29.            IOUtils.closeStream(fis);
  30.        }

  1. 合并文件

    在window命令窗口中执行

    type A2 >> A1 然后更改后缀为rar即可

14.一致性模型

  1. @Test
  2. 	public void writeFile() throws Exception{
  3. 		// 1 创建配置信息对象
  4. 		Configuration configuration = new Configuration();
  5. 		fs = FileSystem.get(configuration);
  6. 		
  7. 		// 2 创建文件输出流
  8. 		Path path = new Path("F:\\date\\H.txt");
  9. 		FSDataOutputStream fos = fs.create(path);
  10. 		
  11. 		// 3 写数据
  12. 		fos.write("hello Andy".getBytes());
  13.         // 4 一致性刷新
  14. 		fos.hflush();
  15. 		
  16. 		fos.close();
  17. 	}

总结：

写入数据时，如果希望数据被其他client立即可见，调用如下方法

FSDataOutputStream. hflush (); //清理客户端缓冲区数据，被其他client立即可见

内容来源于网络，如有侵权，请联系作者删除！

hdfs

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com