devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

HDFS Java API编程_hdfs java api编程 ——文件读写

作者：笔触狂放9 | 2024-06-15 10:02:22

赞

踩

hdfs java api编程 ——文件读写

第1关：文件读写

知识点

1.HDFS文件创建和操作步骤

step1：获取FileSystem对象；

step2：通过FSDataOutputStream进行写入；

step3：通过FSDataInputStream将文件内容输出。

编程要求

  • 获取hadoop的系统设置，并在其中创建HDFS文件，文件路径为/user/hadoop/myfile；
  • 在myfile文件中添加字符串https://www.educoder.net；
  • 读取刚刚创建myfile文件中的内容，并输出。
  1. import java.io.*;
  2. import java.sql.Date;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.FSDataInputStream;
  5. import org.apache.hadoop.fs.FSDataOutputStream;
  6. import org.apache.hadoop.fs.FileStatus;
  7. import org.apache.hadoop.fs.FileSystem;
  8. import org.apache.hadoop.fs.Path;
  9.  
  10.  
  11. public class hdfs {
  12.  
  13.     public static void main(String[] args) throws IOException {
  14. 		//请在 Begin-End 之间添加代码，完成任务要求。
  15.         /********* Begin *********/
  16.         //获取FileSystem对象
  17.         Configuration conf=new Configuration();  //实现hadoop各模块之间值的传递
  18.         FileSystem fs=FileSystem.get(conf); //获取文件系统
  19.         Path file=new Path("/user/hadoop/myfile");   //创建文件
  20.         //通过FSDataOutputStream进行写入
  21.         FSDataOutputStream outStream=fs.create(file);  //获取输出流
  22.         outStream.writeUTF("https://www.educoder.net"); //可写入任意字符
  23.         outStream.close();  //记得关闭输出流
  24.         //通过FSDataInputStream将文件内容输出
  25.         FSDataInputStream inStream=fs.open(file);  //获取输入流
  26.         String data=inStream.readUTF(); //读取文件
  27.  
  28.         		/********* End *********/
  29.  
  30.     }
  31.   }
  32.

第2关：文件上传

编程要求

完成向HDFS中上传文本文件，如果指定的文件在HDFS中已经存在，由用户指定是追加到原有文件末尾还是覆盖原有的文件。

  1. import java.io.*;
  2. import java.sql.Date;
  3. import java.util.Scanner;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.FSDataInputStream;
  6. import org.apache.hadoop.fs.FSDataOutputStream;
  7. import org.apache.hadoop.fs.FileStatus;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10.  
  11.  
  12. public class hdfs {
  13.  
  14.       /**
  15.      * 判断路径是否存在
  16.      */
  17.     public static boolean test(Configuration conf, String path) throws IOException {
  18.        	/*****start*****/
  19.         //请在此处编写判断文件是否存在的代码
  20.         FileSystem fs = FileSystem.get(conf);//获取对象
  21.         fs.exists(new Path(path));    //判断该路径的文件是否存在，是则返回true
  22.         return fs.exists(new Path(path));
  23.         
  24.         /*****end*****/
  25.     }
  26.  
  27.     /**
  28.      * 复制文件到指定路径
  29.      * 若路径已存在，则进行覆盖
  30.      */
  31.     public static void copyFromLocalFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
  32.         /*****start*****/
  33.         //请在此处编写复制文件到指定路径的代码
  34.         FileSystem fs = FileSystem.get(conf);//获取对象
  35.         Path localPath=new Path(localFilePath);
  36.         Path remotePath=new Path(remoteFilePath);
  37.          /* fs.copyFromLocalFile 第一个参数表示是否删除源文件，第二个参数表示是否覆盖 */
  38.         fs.copyFromLocalFile(false,true,localPath,remotePath);
  39.  
  40.         /*****end*****/
  41.     }
  42.  
  43.     /**
  44.      * 追加文件内容
  45.      */
  46.     public static void appendToFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
  47.        /*****start*****/
  48.         //请在此处编写追加文件内容的代码
  49.         FileSystem fs=FileSystem.get(conf);
  50.         Path remotePath=new Path(remoteFilePath);
  51.         //创建一个文件读入流
  52.         FileInputStream in=new FileInputStream(localFilePath);
  53.         //创建一个文件输出流。输出内容将追加到文件末尾
  54.         FSDataOutputStream out=fs.append(remotePath);
  55.         //读写文件内容
  56.         byte[] data=new byte[1024];
  57.         int read=-1;
  58.         while((read=in.read(data))>0){
  59.             out.write(data,0,read);
  60.         }
  61.    
  62.         /*****end*****/
  63.     }
  64.     
  65. 	/**
  66. 	 * 主函数
  67. 	 */
  68. 	public static void main(String[] args)throws IOException  {
  69. 		Configuration conf = new Configuration();
  70.   
  71.         createHDFSFile(conf);
  72.      
  73. 		String localFilePath = "./file/text.txt";			// 本地路径
  74. 		String remoteFilePath = "/user/hadoop/text.txt";    // HDFS路径
  75. 		String choice = "";    
  76. 		
  77. 		try {
  78. 			/* 判断文件是否存在 */
  79. 			Boolean fileExists = false;
  80. 			if (hdfs.test(conf, remoteFilePath)) {
  81. 				fileExists = true;
  82. 				System.out.println(remoteFilePath + " 已存在.");
  83.                 choice = "append";		//若文件存在则追加到文件末尾
  84. 			} else {
  85. 				System.out.println(remoteFilePath + " 不存在.");
  86.                 choice = "overwrite";	//覆盖
  87. 			}
  88.  
  89.             /*****start*****/
  90.        		//请在此处编写文件不存在则上传 文件choice等于overwrite则覆盖   choice 等于append 则追加的逻辑
  91.         
  92.        		 if (fileExists!=true ) { // 文件不存在，则上传
  93. 				
  94. 				System.out.println(localFilePath + " 已上传至 " + remoteFilePath);
  95. 			} else if (choice=="overwrite" ) {    // 选择覆盖
  96. 				
  97. 				System.out.println(localFilePath + " 已覆盖 " + remoteFilePath);
  98. 			} else if ( choice=="append") {   // 选择追加
  99. 				
  100. 				System.out.println(localFilePath + " 已追加至 " + remoteFilePath);
  101. 			}
  102.         
  103.         
  104.        		/*****end*****/
  105.             
  106. 			
  107. 		} catch (Exception e) {
  108. 			e.printStackTrace();
  109. 		}
  110. 	}
  111.  
  112. 	//创建HDFS文件
  113. 	public static void createHDFSFile(Configuration conf)throws IOException{
  114.         FileSystem fs = FileSystem.get(conf);  //获取文件系统
  115.         Path file = new Path("/user/hadoop/text.txt");        //创建文件   
  116.         FSDataOutputStream outStream = fs.create(file); //获取输出流
  117.         outStream.writeUTF("hello##Hdfs");
  118.         outStream.close();
  119.         fs.close();
  120.     }
  121.    
  122.  
  123. }

第3关：文件下载

编程要求

完成从HDFS中下载文件的功能

  1. import java.io.*;
  2. import java.sql.Date;
  3. import java.util.Scanner;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.FSDataInputStream;
  6. import org.apache.hadoop.fs.FSDataOutputStream;
  7. import org.apache.hadoop.fs.FileStatus;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10.  
  11.  
  12. public class hdfs {
  13. 	 /**
  14.      * 下载文件到本地
  15.      * 判断本地路径是否已存在，若已存在，则自动进行重命名
  16.      */
  17.     public static void copyToLocal(Configuration conf, String remoteFilePath, String localFilePath) throws IOException {
  18.         FileSystem fs = FileSystem.get(conf);
  19.         Path remotePath = new Path(remoteFilePath);
  20. 		File f = new File(localFilePath);
  21. 		/*****start*****/
  22.         /*在此添加判断文件是否存在的代码，如果文件名存在，自动重命名(在文件名后面加上 _0, _1 ...) */
  23.         if (fs.exists(new Path(localFilePath))) {
  24.         	System.out.println(localFilePath + " 已存在.");
  25.         	Integer i = 0;
  26.         	while ( f.exists() ) {
  27.         		f = new File( localFilePath     );
  28.         		if ( f.exists() ) {
  29.         			localFilePath = localFilePath + "_" +  i    ;
  30.         			break;
  31.         		}
  32.         	}
  33.         	System.out.println("将重新命名为: " + localFilePath);
  34.         }
  35.  
  36.         /*****end*****/
  37.  
  38. 		/*****start*****/
  39.         // 在此添加将文件下载到本地的代码
  40. 		Path localPath=new Path(localFilePath);
  41.         fs.copyToLocalFile(remotePath,localPath);
  42.  
  43. 	   /*****end*****/
  44.        fs.close();
  45.     }
  46.     
  47. 	/**
  48. 	 * 主函数
  49. 	 */
  50. 	public static void main(String[] args)throws IOException {
  51. 		Configuration conf = new Configuration();
  52. 		createHDFSFile(conf);
  53. 		String localFilePath = "/tmp/output/text.txt";    // 本地路径
  54. 		String remoteFilePath = "/user/hadoop/text.txt";    // HDFS路径
  55. 		
  56. 		try {
  57. 			//调用方法下载文件至本地
  58. 			hdfs.copyToLocal(conf, remoteFilePath, localFilePath);
  59. 			System.out.println("下载完成");
  60. 		} catch (Exception e) {
  61. 			e.printStackTrace();
  62. 		}
  63. 	}
  64.  
  65. 	
  66. 	//创建HDFS文件
  67. 	public static void createHDFSFile(Configuration conf)throws IOException{
  68.         FileSystem fs = FileSystem.get(conf);  //获取文件系统
  69.         Path file = new Path("/user/hadoop/text.txt");        //创建文件   
  70.         FSDataOutputStream outStream = fs.create(file); //获取输出流
  71.         outStream.writeUTF("hello hadoop HDFS www.educoder.net");
  72.         outStream.close();
  73.         fs.close();
  74.     }
  75.  
  76. }

第4关：使用字符流读取数据

知识点

1.使用字符流读取数据分为三步：

step1：通过Configuration对象获取FileSystem对象；

step2：通过fs获取FSDataInputStream对象；

step3：通过字符流循环读取文件中数据并输出。

编程要求

完成将HDFS中指定文件输出到指定文件中

  1. import java.io.*;
  2. import java.sql.Date;
  3. import java.util.Scanner;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.FSDataInputStream;
  6. import org.apache.hadoop.fs.FSDataOutputStream;
  7. import org.apache.hadoop.fs.FileStatus;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10.  
  11.  
  12. public class hdfs {
  13.  /**
  14.      * 读取文件内容
  15.      */
  16.     public static void cat(Configuration conf, String remoteFilePath) throws IOException {
  17.        
  18. 		
  19. 		/*****start*****/
  20. 		//1.读取文件中的数据
  21.         FileSystem fs=FileSystem.get(conf);
  22.         Path remotePath=new Path(remoteFilePath);
  23.         FSDataInputStream in=fs.open(remotePath);
  24.         BufferedReader d=new BufferedReader(new InputStreamReader(in));
  25.         StringBuffer buffer=new StringBuffer();
  26.         String line=null;
  27.         while((line=d.readLine())!=null)
  28.         {
  29.             buffer.append(line);
  30.         }
  31.         String res = buffer.toString();
  32.  
  33.  
  34. 		//2.将读取到的数据输出到  /tmp/output/text.txt 文件中  提示：可以使用FileWriter
  35.         FileWriter f1=new FileWriter("/tmp/output/text.txt");
  36.         f1.write(res);
  37.         f1.close();
  38. 	
  39.  
  40. 	   /*****end*****/
  41.     }
  42.     
  43. 	/**
  44. 	 * 主函数
  45. 	 */
  46. 	public static void main(String[] args)throws IOException {
  47. 		Configuration conf = new Configuration();
  48. 		createHDFSFile(conf);
  49. 		String remoteFilePath = "/user/hadoop/text.txt";    // HDFS路径
  50. 		
  51. 		try {
  52. 			System.out.println("读取文件: " + remoteFilePath);
  53. 			hdfs.cat(conf, remoteFilePath);
  54. 			System.out.println("\n读取完成");
  55. 		} catch (Exception e) {
  56. 			e.printStackTrace();
  57. 		}
  58. 	}
  59.  
  60. 	//创建HDFS文件
  61. 	public static void createHDFSFile(Configuration conf)throws IOException{
  62.         FileSystem fs = FileSystem.get(conf);  //获取文件系统
  63.         Path file = new Path("/user/hadoop/text.txt");        //创建文件   
  64.         FSDataOutputStream outStream = fs.create(file); //获取输出流
  65.         outStream.writeUTF("hello hadoop HDFS step4 www.educoder.net");
  66.         outStream.close();
  67.         fs.close();
  68.     }
  69.  
  70.  
  71. }

第5关：删除文件

知识点

1.public boolean delete(Path f, Boolean recursive) 永久性删除指定的文件或目录，如果f是一个空目录或者文件，那么recursive的值就会被忽略。只有recursive＝true时，一个非空目录及其内容才会被删除（即递归删除所有文件）。

编程要求

删除HDFS中/user/hadoop/text.txt文件

  1. import java.io.*;
  2. import java.sql.Date;
  3. import java.util.Scanner;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.FSDataInputStream;
  6. import org.apache.hadoop.fs.FSDataOutputStream;
  7. import org.apache.hadoop.fs.FileStatus;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10.  
  11.  
  12. public class hdfs {
  13.  
  14. 	 /**
  15.      * 删除文件 
  16.      */
  17.     public static boolean rm(Configuration conf, String remoteFilePath) throws IOException {
  18. 		/*****start*****/        
  19. 		//请在此添加删除文件的代码
  20.         FileSystem fs=FileSystem.get(conf);
  21.         Path remotePath=new Path(remoteFilePath);
  22.         boolean result=fs.delete(remotePath,false);
  23.  
  24. 		return   result     ;
  25.  
  26.  
  27. 		/*****end*****/
  28.     }
  29.     
  30. 	/**
  31. 	 * 主函数
  32. 	 */
  33. 	public static void main(String[] args) {
  34. 		Configuration conf = new Configuration();
  35. 		String remoteFilePath = "/user/hadoop/text.txt";    // HDFS文件
  36. 		
  37. 		try {
  38. 			if (rm(conf, remoteFilePath) ) {
  39. 				System.out.println("文件删除: " + remoteFilePath);
  40. 			} else {
  41. 				System.out.println("操作失败（文件不存在或删除失败）");
  42. 			}
  43. 		} catch (Exception e) {
  44. 			e.printStackTrace();
  45. 		}
  46. 	}
  47.  
  48.  
  49. }

第6关：删除文件夹

知识点

1.验证目录下是否存在文件：public RemoteIterator<LocatedFileStatus> listFiles(Path f, Boolean recursive)，recursive是false，则返回目录中的文件；如果recursive是true，则在根目录中返回文件。

2.删除HDFS中的文件和目录：public boolean delete(Path f, Boolean recursive)

编程要求

删除HDFS中/user/hadoop/tmp目录和/user/hadoop/dir目录，删除前，需要判断两个目录是否为空，若不为空则不删除，否则删除。

  1. import java.io.*;
  2. import java.sql.Date;
  3. import java.util.Scanner;
  4. import org.apache.hadoop.conf.Configuration;
  5. import org.apache.hadoop.fs.*;
  6.  
  7.  
  8. public class hdfs {
  9.  
  10. 	
  11.     /**
  12.      * 判断目录是否为空
  13.      * true: 空，false: 非空
  14.      */
  15.     public static boolean isDirEmpty(Configuration conf, String remoteDir) throws IOException {
  16.        /*****start*****/
  17. 		//请在此添加判断目录是否为空的代码
  18.         FileSystem fs=FileSystem.get(conf);
  19.         Path dirPath=new Path(remoteDir);
  20.         RemoteIterator<LocatedFileStatus> result=fs.listFiles(dirPath,true);
  21.  
  22. 		return !result.hasNext() ;
  23. 		/*****end*****/
  24.     }
  25. 	
  26.     /**
  27.      * 删除目录
  28.      */
  29.     public static boolean rmDir(Configuration conf, String remoteDir, boolean recursive) throws IOException {
  30.        /*****start*****/
  31. 		//请在此添加删除目录的代码
  32.         FileSystem fs=FileSystem.get(conf);
  33.         Path dirPath=new Path(remoteDir);
  34.         boolean result=fs.delete(dirPath,false);
  35. 		return result ;
  36. 	   /*****end*****/
  37.  
  38.     }
  39.     
  40. 	/**
  41. 	 * 主函数
  42. 	 */
  43. 	public static void main(String[] args) {
  44. 		Configuration conf = new Configuration();
  45. 		String remoteDir = "/user/hadoop/dir/";    // HDFS目录
  46. 		String remoteDir1 = "/user/hadoop/tmp/";    // HDFS目录
  47. 		Boolean forceDelete = false;  // 是否强制删除
  48. 		
  49. 		try {
  50. 			if ( !isDirEmpty(conf, remoteDir) && !forceDelete ) {
  51. 				System.out.println("目录不为空，不删除");
  52. 			} else {
  53. 				if ( rmDir(conf, remoteDir, forceDelete) ) {
  54. 					System.out.println("目录已删除: " + remoteDir);
  55. 				} else {
  56. 					System.out.println("操作失败");
  57. 				}
  58. 			}
  59.             
  60.             if ( !isDirEmpty(conf, remoteDir1) && !forceDelete ) {
  61. 				System.out.println("目录不为空，不删除");
  62. 			} else {
  63. 				if ( rmDir(conf, remoteDir1, forceDelete) ) {
  64. 					System.out.println("目录已删除: " + remoteDir1);
  65. 				} else {
  66. 					System.out.println("操作失败");
  67. 				}
  68. 			}
  69. 		} catch (Exception e) {
  70. 			e.printStackTrace();
  71. 		}
  72. 	}
  73.  
  74.  
  75. }

第7关：自定义数据输入流

知识点

1.相关方法

public int read(char[] cbuf,int off,int len)throws IOException

（1）已经读取了指定的字符数， 底层流的 read 方法返回 -1，指示文件末尾（end-of-file），或者底层流的 ready 方法返回 false，指示将阻塞后续的输入请求。

（2） 如果第一次对底层流调用 read 返回 -1（指示文件末尾），则此方法返回 -1，否则此方法返回实际读取的字符数。

编程要求

实现按行读取HDFS中指定文件的方法readLine()，如果读到文件末尾，则返回空，否则返回文件一行的文本，即实现和BufferedReader类的readLine()方法类似的效果。

  1. import org.apache.hadoop.conf.Configuration;
  2. import org.apache.hadoop.fs.FSDataInputStream;
  3. import org.apache.hadoop.fs.FileSystem;
  4. import org.apache.hadoop.fs.Path;
  5. import java.io.*;
  6.  
  7. public class MyFSDataInputStream extends FSDataInputStream {
  8. 	public MyFSDataInputStream(InputStream in) {
  9. 		super(in);
  10. 	}
  11. 	
  12. 	/**
  13.      * 实现按行读取
  14.      * 每次读入一个字符，遇到"\n"结束，返回一行内容
  15.      */
  16. 	public static String readline(BufferedReader br) throws IOException {
  17. 	/*****start*****/
  18. 	//请在此处填充代码实现相关功能
  19.         char[] data = new char[1024];
  20.         int read = -1;
  21.         int off = 0; // 循环执行时，br 每次会从上一次读取结束的位置继续读取，因此该函数里，off 每次都从0开始
  22.         while ( (read = br.read(data, off, 1)) != -1 ) {
  23.             if (String.valueOf(data[off]).equals("\n") ) {
  24.                 off += 1;
  25.                 return String.valueOf(data, 0, read);
  26.             }
  27.             off += 1;
  28.             return String.valueOf(data, 0, read);
  29.         }
  30.     
  31. 	return  null   ;
  32. 	/*****end*****/
  33.  
  34. 	}
  35.  
  36. 	/**
  37.      * 读取文件内容
  38.      */
  39.     public static void cat(Configuration conf, String remoteFilePath) throws IOException {
  40.         FileSystem fs = FileSystem.get(conf);
  41.         Path remotePath = new Path(remoteFilePath);
  42.         FSDataInputStream in = fs.open(remotePath);
  43.         BufferedReader br = new BufferedReader(new InputStreamReader(in));
  44.         FileWriter f = new FileWriter("/tmp/output/text.txt");
  45.         String line = null;
  46.         while ( (line = MyFSDataInputStream.readline(br)) != null ) {
  47.         	f.write(line);
  48.         }
  49.         f.close();
  50.         br.close();
  51.         in.close();
  52.         fs.close();
  53.     }
  54. 	
  55. 	/**
  56. 	 * 主函数
  57. 	 */
  58. 	public static void main(String[] args) {
  59. 		Configuration conf = new Configuration();
  60. 		String remoteFilePath = "/user/hadoop/text.txt";    // HDFS路径
  61. 		try {
  62. 			MyFSDataInputStream.cat(conf, remoteFilePath);
  63. 		} catch (Exception e) {
  64. 			e.printStackTrace();
  65. 		}
  66. 	}
  67. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/721864
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号