HDFS源码分析EditLog之获取编辑日志输入流

2023-01-31 19:42:08

在《HDFS源码分析之EditLogTailer》一文中，我们详细了解了编辑日志跟踪器EditLogTailer的实现，介绍了其内部编辑日志追踪线程EditLogTailerThread的实现，及其线程完成编辑日志跟踪所依赖的最重要的方法，执行日志追踪的doTailEdits()方法。在该方法的处理流程中，首先需要从编辑日志editLog中获取编辑日志输入流集合streams，获取的输入流为最新事务ID加1之后的数据。那么这个编辑日志输入流集合streams是如何获取的呢？本文我们将进行详细研究。

在doTailEdits()方法中，获取编辑日志输入流的代码如下：

  1. // 从编辑日志editLog中获取编辑日志输入流集合streams，获取的输入流为最新事务ID加1之后的数据
  2. streams = editLog.selectInputStreams(lastTxnId + 1, 0, null, false);

这个editLog便是一个文件系统编辑日志FSEditLog实例，我们看下它的selectInputStreams()方法，代码如下：

  1. /**
  2. * Select a list of input streams.
  3. *
  4. * @param fromTxId first transaction in the selected streams
  5. * @param toAtLeastTxId the selected streams must contain this transaction
  6. * @param recovery recovery context
  7. * @param inProgressOk set to true if in-progress streams are OK
  8. */
  9. public Collection<EditLogInputStream> selectInputStreams(
  10. long fromTxId, long toAtLeastTxId, MetaRecoveryContext recovery,
  11. boolean inProgressOk) throws IOException {
  12. / 创建编辑日志输入流EditLogInputStream列表streams
  13. List<EditLogInputStream> streams = new ArrayList<EditLogInputStream>();
  14. // 在Object对象journalSetLock上使用synchronized进行同步
  15. synchronized(journalSetLock) {
  16. // 检测journalSet状态
  17. Preconditions.checkState(journalSet.isOpen(), "Cannot call " +
  18. "selectInputStreams() on closed FSEditLog");
  19. // 调用三个参数的selectInputStreams()方法，传入空的streams列表，从fromTxId事务ID开始，
  20. // 编辑日志同步时，标志位inProgressOk为false
  21. selectInputStreams(streams, fromTxId, inProgressOk);
  22. }
  23. try {
  24. // 数据监测
  25. checkForGaps(streams, fromTxId, toAtLeastTxId, inProgressOk);
  26. } catch (IOException e) {
  27. if (recovery != null) {
  28. // If recovery mode is enabled, continue loading even if we know we
  29. // can't load up to toAtLeastTxId.
  30. LOG.error(e);
  31. } else {
  32. closeAllStreams(streams);
  33. throw e;
  34. }
  35. }
  36. return streams;
  37. }

它首先会创建编辑日志输入流EditLogInputStream列表streams，并在Object对象journalSetLock上使用synchronized进行同步，检测journalSet状态，随后调用三个参数的selectInputStreams()方法，传入空的streams列表，从fromTxId事务ID开始，编辑日志是否可处于处理过程中的标志位inProgressOk，编辑日志同步时，标志位inProgressOk为false，最后，调用checkForGaps()方法进行相关数据监测。

我们继续看下三个参数的selectInputStreams()方法，代码如下：

  1. @Override
  2. public void selectInputStreams(Collection<EditLogInputStream> streams,
  3. long fromTxId, boolean inProgressOk) throws IOException {
  4. / 调用JournalSet的同名方法
  5. journalSet.selectInputStreams(streams, fromTxId, inProgressOk);
  6. }

它其实是调用JournalSet的同名方法。JournalSet是什么呢？它是Journal集合的管理者，而Journal就是日志的意思，它是Hadoop HA中EditLog在JournalNode上的组织形式。我们看下JournalSet的selectInputStreams()方法，代码如下：

  1. /**
  2. * In this function, we get a bunch of streams from all of our JournalManager
  3. * objects.  Then we add these to the collection one by one.
  4. * 在这个方法内，我们从所有JournalManager对象中得到一堆输入流。接着我们把它们一个接一个的添加到集合中。
  5. *
  6. * @param streams          The collection to add the streams to.  It may or
  7. *                         may not be sorted-- this is up to the caller.
  8. * @param fromTxId         The transaction ID to start looking for streams at
  9. * @param inProgressOk     Should we consider unfinalized streams?
  10. */
  11. @Override
  12. public void selectInputStreams(Collection<EditLogInputStream> streams,
  13. long fromTxId, boolean inProgressOk) throws IOException {
  14. final PriorityQueue<EditLogInputStream> allStreams =
  15. new PriorityQueue<EditLogInputStream>(64,
  16. EDIT_LOG_INPUT_STREAM_COMPARATOR);
  17. // 遍历journals，得到每个JournalAndStream实例jas
  18. for (JournalAndStream jas : journals) {
  19. // 如果jas不可用，记录日志，跳过
  20. if (jas.isDisabled()) {
  21. LOG.info("Skipping jas " + jas + " since it's disabled");
  22. continue;
  23. }
  24. // 利用jas得到JournalManager实例，然后调用其selectInputStreams()方法，获得输入流，并放入allStreams输入流集合
  25. try {
  26. jas.getManager().selectInputStreams(allStreams, fromTxId, inProgressOk);
  27. } catch (IOException ioe) {
  28. LOG.warn("Unable to determine input streams from " + jas.getManager() +
  29. ". Skipping.", ioe);
  30. }
  31. }
  32. // 过滤掉无用的编辑日志输入流
  33. chainAndMakeRedundantStreams(streams, allStreams, fromTxId);
  34. }

这个方法的大体处理流程如下：

1、遍历JournalSet中的journals，得到每个JournalAndStream实例jas：

1.1、如果jas不可用，记录日志，跳过；

1.2、利用jas得到JournalManager实例，然后调用其selectInputStreams()方法，获得输入流，并放入allStreams输入流集合；

2、调用chainAndMakeRedundantStreams()方法，过滤掉无用的编辑日志输入流。
        首先，我们来看下JournalManager实例的selectInputStreams()方法，我们以FileJournalManager为例，代码如下：

  1. @Override
  2. synchronized public void selectInputStreams(
  3. Collection<EditLogInputStream> streams, long fromTxId,
  4. boolean inProgressOk) throws IOException {
  5. // 先调用StorageDirectory的getCurrentDir()方法获得其current目录，
  6. // 然后再调用matchEditLogs()方法，获得编辑日志文件EditLogFile列表elfs
  7. List<EditLogFile> elfs = matchEditLogs(sd.getCurrentDir());
  8. LOG.debug(this + ": selecting input streams starting at " + fromTxId +
  9. (inProgressOk ? " (inProgress ok) " : " (excluding inProgress) ") +
  10. "from among " + elfs.size() + " candidate file(s)");
  11. // 调用addStreamsToCollectionFromFiles()方法，根据编辑日志文件列表elfs获得输入流，并添加到输入流列表streams
  12. addStreamsToCollectionFromFiles(elfs, streams, fromTxId, inProgressOk);
  13. }

这个方法的处理流程如下：

1、先调用StorageDirectory的getCurrentDir()方法获得其current目录，然后再调用matchEditLogs()方法，获得编辑日志文件EditLogFile列表elfs；

2、调用addStreamsToCollectionFromFiles()方法，根据编辑日志文件列表elfs获得输入流，并添加到输入流列表streams。

我们再看下matchEditLogs()方法，代码如下：

  1. /**
  2. * returns matching edit logs via the log directory. Simple helper function
  3. * that lists the files in the logDir and calls matchEditLogs(File[])
  4. *
  5. * @param logDir
  6. *          directory to match edit logs in
  7. * @return matched edit logs
  8. * @throws IOException
  9. *           IOException thrown for invalid logDir
  10. */
  11. public static List<EditLogFile> matchEditLogs(File logDir) throws IOException {
  12. return matchEditLogs(FileUtil.listFiles(logDir));
  13. }
  14. static List<EditLogFile> matchEditLogs(File[] filesInStorage) {
  15. return matchEditLogs(filesInStorage, false);
  16. }
  17. private static List<EditLogFile> matchEditLogs(File[] filesInStorage,
  18. boolean forPurging) {
  19. // 创建编辑日志文件EditLogFile列表ret
  20. List<EditLogFile> ret = Lists.newArrayList();
  21. // 遍历filesInStorage，对每个文件进行处理
  22. for (File f : filesInStorage) {
  23. // 获得文件名
  24. String name = f.getName();
  25. // Check for edits
  26. // 根据文件名，利用正则表达式，检测其是否为编辑日志edits log
  27. // 正则表达式为edits_(\d+)-(\d+)
  28. // 原edits_(\\d+)-(\\d+)中d前面的两个\中第一个只是转义符
  29. // 文件名类似如下：edits_0000000000001833048-0000000000001833081
  30. // 第一串数字为起始事务ID，第二串数字为终止事务ID
  31. Matcher editsMatch = EDITS_REGEX.matcher(name);
  32. if (editsMatch.matches()) {// 文件名匹配正则表达式的话，说明其是我们需要寻找的编辑日志文件
  33. try {
  34. // 获取起始事务ID：正则表达式中第一个匹配的是起始事务ID
  35. long startTxId = Long.parseLong(editsMatch.group(1));
  36. // 获取终止事务ID：正则表达式中第二个匹配的是终止事务ID
  37. long endTxId = Long.parseLong(editsMatch.group(2));
  38. // 利用文件f、起始事务艾迪startTxId、终止事务艾迪endTxId构造编辑日志文件EditLogFile实例，
  39. // 并添加到ret列表中
  40. ret.add(new EditLogFile(f, startTxId, endTxId));
  41. continue;
  42. } catch (NumberFormatException nfe) {
  43. LOG.error("Edits file " + f + " has improperly formatted " +
  44. "transaction ID");
  45. // skip
  46. }
  47. }
  48. // Check for in-progress edits
  49. // 检测正在处理中的编辑日志
  50. // 正则表达式为edits_inprogress_(\d+)
  51. // 原edits_inprogress_(\\d+)中d前面的两个\中第一个只是转义符
  52. Matcher inProgressEditsMatch = EDITS_INPROGRESS_REGEX.matcher(name);
  53. if (inProgressEditsMatch.matches()) {
  54. try {
  55. long startTxId = Long.parseLong(inProgressEditsMatch.group(1));
  56. ret.add(
  57. new EditLogFile(f, startTxId, HdfsConstants.INVALID_TXID, true));
  58. continue;
  59. } catch (NumberFormatException nfe) {
  60. LOG.error("In-progress edits file " + f + " has improperly " +
  61. "formatted transaction ID");
  62. // skip
  63. }
  64. }
  65. if (forPurging) {
  66. // Check for in-progress stale edits
  67. Matcher staleInprogressEditsMatch = EDITS_INPROGRESS_STALE_REGEX
  68. .matcher(name);
  69. if (staleInprogressEditsMatch.matches()) {
  70. try {
  71. long startTxId = Long.valueOf(staleInprogressEditsMatch.group(1));
  72. ret.add(new EditLogFile(f, startTxId, HdfsConstants.INVALID_TXID,
  73. true));
  74. continue;
  75. } catch (NumberFormatException nfe) {
  76. LOG.error("In-progress stale edits file " + f + " has improperly "
  77. + "formatted transaction ID");
  78. // skip
  79. }
  80. }
  81. }
  82. }
  83. return ret;
  84. }

我们看下最核心的两个参数的matchEditLogs()方法，它的处理流程为：

1、创建编辑日志文件EditLogFile列表ret；

2、遍历filesInStorage，对每个文件进行处理：

2.1、获得文件名name；

2.2、根据文件名name，利用正则表达式，检测其是否为编辑日志edits log：

利用的正则表达式为edits_(\d+)-(\d+)，原edits_(\\d+)-(\\d+)中d前面的两个\中第一个只是转义符，文件名类似如下：edits_0000000000001833048-0000000000001833081，第一串数字为起始事务ID，第二串数字为终止事务ID；

2.3、文件名匹配正则表达式的话，说明其是我们需要寻找的编辑日志文件：

2.3.1、获取起始事务艾迪startTxId：正则表达式中第一个匹配的是起始事务ID；

2.3.2、获取终止事务艾迪endTxId：正则表达式中第二个匹配的是终止事务ID；

2.3.3、利用文件f、起始事务艾迪startTxId、终止事务艾迪endTxId构造编辑日志文件EditLogFile实例，并添加到ret列表中；

2.3.4、continue，继续循环下一个文件；

2.4、根据文件名name，利用正则表达式，检测其是否为正在处理中的编辑日志edits log：

利用的正则表达式为edits_inprogress_(\d+)，原edits_inprogress_(\\d+)中d前面的两个\中第一个只是转义符，文件名类似如下：edits_inprogress_0000000000001853186，后面的字符串为起始事务ID；

2.5、文件名匹配正则表达式的话，说明其是我们需要寻找的正在处理中的编辑日志文件：

2.5.1、获取起始事务艾迪startTxId：正则表达式中第一个匹配的是起始事务ID；

2.5.2、利用文件f、起始事务艾迪startTxId、终止事务艾迪-12345构造编辑日志文件EditLogFile实例，并添加到ret列表中；

2.5.3、continue，继续循环下一个文件；

3、返回编辑日志文件EditLogFile列表ret。

再回到FileJournalManager的selectInputStreams()方法，我们看下它的第二步：调用addStreamsToCollectionFromFiles()方法，根据编辑日志文件列表elfs添加输入流列表streams，代码如下：

  1. static void addStreamsToCollectionFromFiles(Collection<EditLogFile> elfs,
  2. Collection<EditLogInputStream> streams, long fromTxId, boolean inProgressOk) {
  3. // 遍历EditLogFile集合elfs，针对每个EditLogFile实例elf进行如下处理：
  4. for (EditLogFile elf : elfs) {
  5. // 如果elf处于处理过程中：
  6. if (elf.isInProgress()) {
  7. // 如果不需要获取处于处理过程中的编辑日志，直接跳过
  8. if (!inProgressOk) {
  9. LOG.debug("passing over " + elf + " because it is in progress " +
  10. "and we are ignoring in-progress logs.");
  11. continue;
  12. }
  13. // 校验编辑日志，校验不成功的话，直接跳过
  14. try {
  15. elf.validateLog();
  16. } catch (IOException e) {
  17. LOG.error("got IOException while trying to validate header of " +
  18. elf + ".  Skipping.", e);
  19. continue;
  20. }
  21. }
  22. // 如果elf的最后事务艾迪lastTxId小于我们获取编辑日志的起始事务艾迪fromTxId，直接跳过
  23. if (elf.lastTxId < fromTxId) {
  24. assert elf.lastTxId != HdfsConstants.INVALID_TXID;
  25. LOG.debug("passing over " + elf + " because it ends at " +
  26. elf.lastTxId + ", but we only care about transactions " +
  27. "as new as " + fromTxId);
  28. continue;
  29. }
  30. // 利用elf中的文件file、起始事务艾迪firstTxId、终止事务艾迪lastTxId、编辑日志文件是否处于处理过程中标志位isInProgress，
  31. // 构造编辑日志文件输入流EditLogFileInputStream实例elfis
  32. EditLogFileInputStream elfis = new EditLogFileInputStream(elf.getFile(),
  33. elf.getFirstTxId(), elf.getLastTxId(), elf.isInProgress());
  34. LOG.debug("selecting edit log stream " + elf);
  35. // 将elfis加入到输入流列表streams
  36. streams.add(elfis);
  37. }
  38. }

它的处理流程如下：

遍历EditLogFile集合elfs，针对每个EditLogFile实例elf进行如下处理：

1、如果elf处于处理过程中，同时如果不需要获取处于处理过程中的编辑日志，直接跳过，否则校验编辑日志，校验不成功的话，直接跳过，成功则继续；

2、如果elf的最后事务艾迪lastTxId小于我们获取编辑日志的起始事务艾迪fromTxId，直接跳过，否则继续；

3、利用elf中的文件file、起始事务艾迪firstTxId、终止事务艾迪lastTxId、编辑日志文件是否处于处理过程中标志位isInProgress，构造编辑日志文件输入流EditLogFileInputStream实例elfis；

4、将elfis加入到输入流列表streams。

上一篇：HDFS源码分析心跳汇报之数据块增量汇报


下一篇：HDFS源码分析EditLog之读取操作符