正文

Java读写HDFS文件

IP云
IP云V管理员/9 阅读
06/27
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

一、依赖包maven路径

  1. <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
  2. <dependency>
  3. <groupId>org.apache.hadoop</groupId>
  4. <artifactId>hadoop-client</artifactId>
  5. <version>2.7.3</version>
  6. <scope>runtime</scope>
  7. </dependency>

二、针对HDFS文件的操作类HDFSOperate

  1. package com.hdfs.util;
  2. import java.io.BufferedReader;
  3. import java.io.File;
  4. import java.io.FileOutputStream;
  5. import java.io.IOException;
  6. import java.io.InputStreamReader;
  7. import java.io.PrintStream;
  8. import java.net.URI;
  9. import org.apache.hadoop.conf.Configuration;
  10. import org.apache.hadoop.fs.FSDataInputStream;
  11. import org.apache.hadoop.fs.FSDataOutputStream;
  12. import org.apache.hadoop.fs.FileSystem;
  13. import org.apache.hadoop.fs.Path;
  14. import org.apache.hadoop.io.IOUtils;
  15. /**
  16. * 针对HDFS文件的操作类
  17. */
  18. public class HDFSOperate {
  19. /**
  20. * 新增（创建）HDFS文件
  21. * @param hdfs
  22. */
  23. public void createHDFS(String hdfs){
  24. try {
  25. Configuration conf = new Configuration();
  26. conf.setBoolean("dfs.support.append", true);
  27. conf.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
  28. conf.set("dfs.client.block.write.replace-datanode-on-failure.enable","true");
  29. FileSystem fs = FileSystem.get(URI.create(hdfs), conf);
  30. Path path = new Path(hdfs);
  31. //判断HDFS文件是否存在
  32. if(fs.exists(path)){
  33. //System.out.println(hdfs + "已经存在！！！");
  34. }else{
  35. FSDataOutputStream hdfsOutStream = fs.create(new Path(hdfs));
  36. hdfsOutStream.close();
  37. }
  38. fs.close();
  39. } catch (Exception e) {
  40. // TODO: handle exception
  41. e.printStackTrace();
  42. }
  43. }
  44. /**
  45. * 在HDFS文件后面追加内容
  46. * @param hdfs
  47. * @param appendContent
  48. */
  49. public void appendHDFS(String hdfs,String appendContent){
  50. try {
  51. Configuration conf = new Configuration();
  52. conf.setBoolean("dfs.support.append", true);
  53. conf.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
  54. conf.set("dfs.client.block.write.replace-datanode-on-failure.enable","true");
  55. FileSystem fs = FileSystem.get(URI.create(hdfs), conf);
  56. Path path = new Path(hdfs);
  57. //判断HDFS文件是否存在
  58. if(fs.exists(path)){
  59. //System.out.println(hdfs + "已经存在！！！");
  60. }else{
  61. FSDataOutputStream hdfsOutStream = fs.create(new Path(hdfs));
  62. hdfsOutStream.close();
  63. }
  64. FSDataOutputStream hdfsOutStream = fs.append(new Path(hdfs));
  65. byte [] str = appendContent.getBytes("UTF-8");//防止中文乱码
  66. hdfsOutStream.write(str);
  67. hdfsOutStream.close();
  68. fs.close();
  69. } catch (Exception e) {
  70. // TODO: handle exception
  71. e.printStackTrace();
  72. }
  73. }
  74. /**
  75. * 修改HDFS文件内容 /  删除就是替换为空
  76. * @param hdfs : hdfs文件路径
  77. * @param sourceContent ：要修改的hdfs文件内容
  78. * @param changeContent ：需要修改成的文件内容
  79. */
  80. public void change(String hdfs,String sourceContent,String changeContent){
  81. try {
  82. Configuration conf = new Configuration();
  83. conf.setBoolean("dfs.support.append", true);
  84. conf.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
  85. conf.set("dfs.client.block.write.replace-datanode-on-failure.enable","true");
  86. FileSystem fs = FileSystem.get(URI.create(hdfs), conf);
  87. Path path = new Path(hdfs);
  88. //判断HDFS文件是否存在
  89. if(fs.exists(path)){
  90. //System.out.println(hdfs + "已经存在！！！");
  91. FSDataInputStream in = fs.open(path);
  92. BufferedReader bf=new BufferedReader(new InputStreamReader(in));//防止中文乱码
  93. String totalString = "";
  94. String line = null;
  95. while ((line = bf.readLine()) != null) {
  96. totalString += line;
  97. }
  98. String changeString = totalString.replace(sourceContent, changeContent);
  99. FSDataOutputStream hdfsOutStream = fs.create(new Path(hdfs));
  100. byte [] str = changeString.getBytes("UTF-8");
  101. hdfsOutStream.write(str);
  102. hdfsOutStream.close();
  103. }else{
  104. //System.out.println(hdfs + "不存在，无需操作！！！");
  105. }
  106. fs.close();
  107. } catch (Exception e) {
  108. // TODO: handle exception
  109. e.printStackTrace();
  110. }
  111. }
  112. /**
  113. * 判断要追加的内容是否存在
  114. * @param hdfs
  115. * @param appendContent
  116. * @return
  117. */
  118. public Boolean isContentExist(String hdfs,String appendContent){
  119. try {
  120. Configuration conf = new Configuration();
  121. conf.setBoolean("dfs.support.append", true);
  122. conf.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
  123. conf.set("dfs.client.block.write.replace-datanode-on-failure.enable","true");
  124. FileSystem fs = FileSystem.get(URI.create(hdfs), conf);
  125. Path path = new Path(hdfs);
  126. //判断HDFS文件是否存在
  127. if(fs.exists(path)){
  128. //System.out.println(hdfs + "已经存在！！！");
  129. FSDataInputStream in = fs.open(path);
  130. BufferedReader bf=new BufferedReader(new InputStreamReader(in));//防止中文乱码
  131. String totalString = "";
  132. String line = null;
  133. while ((line = bf.readLine()) != null) {
  134. totalString += line;
  135. }
  136. if(totalString.contains(appendContent)){
  137. return true;
  138. }
  139. }else{
  140. //System.out.println(hdfs + "不存在，无需操作！！！");
  141. }
  142. fs.close();
  143. } catch (Exception e) {
  144. // TODO: handle exception
  145. e.printStackTrace();
  146. }
  147. return false;
  148. }
  149. public static void main(String[] args) throws IOException {
  150. String hdfs = "hdfs://192.168.168.200:9000/test/tes.txt";
  151. HDFSOperate hdfsOperate = new HDFSOperate();
  152. hdfsOperate.createHDFS(hdfs);
  153. hdfsOperate.appendHDFS(hdfs,"测试新增内容");
  154. //hdfsOperate.change(hdfs, "测试新增内容", "测试修改成功");
  155. }
  156. }