标签：客户 for pre ons inf 自动 发送 ext nod
之前有文章介绍过HDFS租约带来的问题，导致spark应用无法正常读取文件，只能将异常文件找出并且删除后，任务才能继续执行。
但是删除文件实在是下下策，而且文件本身其实并未损坏，只是因为已经close的客户端没有及时的释放租约导致。
按照Hadoop官网的说法，HDFS会启动一个单独的线程，专门处理未及时释放的租约，自动释放超过“硬超时”（默认1小时）仍未释放的租约，但是从问题的现象上来看，这个线程并没有正常的工作，甚至怀疑这个线程是否没有启动，我使用的是CDH集群，可能与相关的设置有关，这一点需要确认。
如果Hadoop没有自动清理租约，我们有办法手动的刷新租约吗？答案是肯定的。
在网上查看资料时，发现HDFS源码中的DistributedFileSystem类提供了一个叫做recoverLease的方法，可以主动的刷新租约。但是非常奇怪，既然已经为外界提供了这个接口，为什么不提供shell指令给用户使用呢？为什么只能通过代码的方式调用呢？我使用的是hadoop-2.6.0，也许后期的版本有所更新，这一点也需要求证。
下面看一下这个方法的源码：
/** * Start the lease recovery of a file * * @param f a file * @return true if the file is already closed * @throws IOException if an error occurs */ public boolean recoverLease(final Path f) throws IOException { Path absF = fixRelativePart(f); return new FileSystemLinkResolver<Boolean>() { @Override public Boolean doCall(final Path p) throws IOException, UnresolvedLinkException { return dfs.recoverLease(getPathName(p)); } @Override public Boolean next(final FileSystem fs, final Path p) throws IOException { if (fs instanceof DistributedFileSystem) { DistributedFileSystem myDfs = (DistributedFileSystem)fs; return myDfs.recoverLease(p); } throw new UnsupportedOperationException("Cannot recoverLease through" + " a symlink to a non-DistributedFileSystem: " + f + " -> " + p); } }.resolve(this, absF); }
有兴趣的朋友可以下载hadoop源码来仔细推敲一下内部的实现原理，这里我们只说如何调用，解决我们的问题：
public static void recoverLease(String path) throws IOException { DistributedFileSystem fs = new DistributedFileSystem(); Configuration conf = new Configuration(); fs.initialize(URI.create(path), conf); fs.recoverLease(new Path(path)); fs.close(); }
这是我编写的一个调用改接口的简单的封装方法，需要注意的是，此处传入的path，必须是包含文件系统以及namenode和端口号的全路径，比如：
hdfs://namenode1:9000/xxx/xxx.log
如果只需要恢复单个文件，调用上述方法即可，但是通常情况下，我们需要对一个目录进行递归的处理，即恢复指定目录下所有租约异常的文件。
这个时候，我们需要先找出指定目录下所有租约异常的文件，形成一个Set或者List，然后再遍历这个容器，对每个文件进行恢复。
寻找文件列表的方法如下：
public static Set<String> getOpenforwriteFileList(String dir) throws IOException { /*拼接URL地址，发送给namenode监听的dfs.namenode.http-address端口，获取所需数据*/ StringBuilder url = new StringBuilder(); url.append("/fsck?ugi=").append("dev"); url.append("&openforwrite=1"); /*获得namenode的主机名以及dfs.namenode.http-address监听端口，例如：http://hadoopnode1:50070*/ Path dirpath; URI namenodeAddress; dirpath = HDFSUtil.getResolvedPath(dir); namenodeAddress = HDFSUtil.getDFSHttpAddress(dirpath); url.insert(0, namenodeAddress); try { url.append("&path=").append(URLEncoder.encode( Path.getPathWithoutSchemeAndAuthority(new Path(dir)).toString(), "UTF-8")); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } Configuration conf = new Configuration(); URLConnectionFactory connectionFactory = URLConnectionFactory.newDefaultURLConnectionFactory(conf); URL path = null; try { path = new URL(url.toString()); } catch (MalformedURLException e) { e.printStackTrace(); } URLConnection connection; BufferedReader input = null; try { connection = connectionFactory.openConnection(path, UserGroupInformation.isSecurityEnabled()); InputStream stream = connection.getInputStream(); input = new BufferedReader(new InputStreamReader(stream, "UTF-8")); } catch (IOException | AuthenticationException e) { e.printStackTrace(); } if (input == null) { System.err.println("Cannot get response from namenode, url = " + url); return null; } String line; Set<String> resultSet = new HashSet<>(); try { while ((line = input.readLine()) != null) { if (line.contains("MISSING") || line.contains("OPENFORWRITE")) { String regEx = "/[^ ]*"; Pattern pattern = Pattern.compile(regEx); Matcher matcher = pattern.matcher(line); while (matcher.find()) { resultSet.add(matcher.group().replaceAll(":", "")); } } } } catch (IOException e) { e.printStackTrace(); } finally { input.close(); } return resultSet; }
其实获取租约异常列表的方法是我从HDFS源码的org.apache.hadoop.hdfs.tools.DFSck中仿照而来的，通过向NameNode的dfs.namenode.http-address端口通信，获取openforwrite状态的文件列表，然后通过正则匹配以及字符串切割，获取所需的内容。
顺便提一句，由于此代码是Java代码，并且返回的Set类型为java.util.Set，如果在Scala代码中调用，则需要将Set类型转化为scala.collection.immutable.Set，具体方法如下：
/*获取需要被恢复租约的文件列表，返回类型为java.util.Set*/ val javaFilesSet = HDFSUtil.getOpenforwriteFileList(hdfsPrefix + recoverDirPath) if (null == javaFilesSet || javaFilesSet.isEmpty) { println("No files need to recover lease : " + hdfsPrefix + recoverDirPath) return } /*将java.util.Set转换成scala.collection.immutable.Set*/ import scala.collection.JavaConverters._ val filesSet = javaFilesSet.asScala.toSet
至此，利用以上两个方法，即可获取指定目录下的所有租约异常的文件列表，然后遍历调用租约恢复接口，即可实现批量恢复。
标签：客户 for pre ons inf 自动 发送 ext nod
原文地址：http://www.cnblogs.com/xyliao/p/7269863.html
