HDFS 读取、写入、遍历目录获取文件全路径、append文件创建或者写入报错问题分析

程序如：

 

1.报错类似下面的信息：

  org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException): Failed to create file [/user/hive/warehouse/test.db/incr_iot_report_gsh_1/odq_changedate=2017-10-19/incr_000000_Meter_0] for [DFSClient_NONMAPREDUCE_1573234124_13]
for client [10.2.65.201], because this file is already being created by [DFSClient_NONMAPREDUCE_-1394763282_13] on [10.2.65.201]
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.recoverLeaseInternal(FSNamesystem.java:3187)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFileInternal(FSNamesystem.java:2968)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFileInt(FSNamesystem.java:3252)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFile(FSNamesystem.java:3216)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.append(NameNodeRpcServer.java:618)
at org.apache.hadoop.hdfs.server.namenode.AuthorizationProviderProxyClientProtocol.append(AuthorizationProviderProxyClientProtocol.java:126)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.append(ClientNamenodeProtocolServerSideTranslatorPB.java:422)
at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)

分析：一般性是多线程造成的文件锁问题，说白点就是一个程序已经打开了这个目录下的这个文件，突然又有一个程序去打开这个文件，才会报如上错误

2.提示文件已经被一个client创建，try again 。。。也是因为文件被多个程序使用，如下错误

org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException): Failed to create file [/user/hive/warehouse/test.db/incr_iot_report_gsh_1/odq_changedate=2017-10-19/incr_000000_Meter_0] for [DFSClient_NONMAPREDUCE_76714100_13]
for client [10.2.65.201], because this file is already being created by [DFSClient_NONMAPREDUCE_1322133037_13] on [10.2.65.201]
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.recoverLeaseInternal(FSNamesystem.java:3187)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFileInternal(FSNamesystem.java:2968)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFileInt(FSNamesystem.java:3252)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFile(FSNamesystem.java:3216)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.append(NameNodeRpcServer.java:618)
at org.apache.hadoop.hdfs.server.namenode.AuthorizationProviderProxyClientProtocol.append(AuthorizationProviderProxyClientProtocol.java:126)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.append(ClientNamenodeProtocolServerSideTranslatorPB.java:422)
at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617)
at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)

处理如上错误可以使用hdfs参数进项控制：代码如下

重新初始化hdfs的句柄

1、从HDFS中读取数据

[java] view
plain copy

Configuration conf = getConf();  

  Path path = new Path(pathstr);   

  FileSystem fs = FileSystem.get(conf);  

   FSDataInputStream fsin= fs.open(path );   

   BufferedReader br =null;  

   String line ;  

   try{  

    br = new BufferedReader(new InputStreamReader(fsin));  

       while ((line = br.readLine()) != null) {  

         System.out.println(line);  

        }   

   }finally{  

    br.close();  

   }  

2、写HDFS

[java] view
plain copy

Configuration conf = getConf();  

Path path = new Path(mid_sort);   

FileSystem fs = FileSystem.get(conf);   

FSDataOutputStream out = fs.create(resultpath);  

out.write(sb.toString().getBytes());  

ou
13ed5
t.close();  

3、遍历目录 获取文件 全路径

[java] view
plain copy

/** 

  * 得到一个目录(不包括子目录)下的所有名字匹配上pattern的文件名 

  * @param fs 

  * @param folderPath 

  * @param pattern 用于匹配文件名的正则 

  * @return 

  * @throws IOException 

  */  

 public static List<Path> getFilesUnderFolder(FileSystem fs, Path folderPath, String pattern) throws IOException {  

  List<Path> paths = new ArrayList<Path>();  

  if (fs.exists(folderPath)) {  

   FileStatus[] fileStatus = fs.listStatus(folderPath);  

   for (int i = 0; i < fileStatus.length; i++) {  

    FileStatus fileStatu = fileStatus[i];  

    if (!fileStatu.isDir()) {//只要文件  

     Path oneFilePath = fileStatu.getPath();  

     if (pattern == null) {  

      paths.add(oneFilePath);  

     } else {  

      if (oneFilePath.getName().contains(pattern)) {  

       paths.add(oneFilePath);  

      }  

     }    

    }  

   }  

  }  

  return paths;  

 }  

4、追加数据 append

[java] view
plain copy

public static boolean appendRTData(String hdfsFile, String appendFile) {  

  boolean flag = false;  

  

  Configuration conf = new Configuration();  

  FileSystem fs = null;  

  try {  

    fs = FileSystem.get(URI.create(hdfsFile), conf);  

    InputStream in = new BufferedInputStream(new FileInputStream(appendFile));  

    OutputStream out = fs.append(new Path(hdfsFile));  

    IOUtils.copyBytes(in, out, 4096, true);  

  } catch (IOException e) {  

    e.printStackTrace();  

  }  

  

  return flag;  

}  

***********************************************************************************************************************************************

***********************************************************************************************************************************************

异常信息

1、Exception in thread "main" java.lang.IllegalArgumentException: java.net.UnknownHostException: ns6

原因是没有加载hdfs的配置信息，需要添加下面的代码：

[java] view
plain copy

conf.addResource(new Path("/xxxx/hdfs-site.xml"));//path是配置文件地址  

如果配置了环境变量可以在不同的机器上使用:

[java] view
plain copy

conf.addResource(new Path(System.getenv("HADOOP_CONF") + "/hdfs-site.xml"));