Hudi Log日志文件读取分析(三)

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: Hudi Log日志文件读取分析(三)

1. 介绍

前面介绍了log日志文件的写入,接着分析log日志文件的读取。

2. 分析

读取日志文件的主要入口为 AbstractHoodieLogRecordScanner#scan,本文分为处理数据块、删除块、控制块来分别讲解其处理流程。

2.1 处理数据块/删除块

在构造 HoodieLogFormatReader后,会通过其 hasNextnext来读取日志文件中的 HoodieLogBlock并处理, scan方法中处理数据块/删除块的核心代码如下

public void scan() {
    HoodieLogFormatReader logFormatReaderWrapper = null;
    try {
      logFormatReaderWrapper = new HoodieLogFormatReader(fs,
          logFilePaths.stream().map(logFile -> new HoodieLogFile(new Path(logFile))).collect(Collectors.toList()),
          readerSchema, readBlocksLazily, reverseReader, bufferSize);
      // 已扫描文件列表
      Set<HoodieLogFile> scannedLogFiles = new HashSet<>();
      while (logFormatReaderWrapper.hasNext()) { // 有下个block
        // 获取日志文件
        HoodieLogFile logFile = logFormatReaderWrapper.getLogFile();
        // 获取下个block块
        HoodieLogBlock r = logFormatReaderWrapper.next();
        if (r.getBlockType() != CORRUPT_BLOCK
            && !HoodieTimeline.compareTimestamps(r.getLogBlockHeader().get(INSTANT_TIME), this.latestInstantTime,
                HoodieTimeline.LESSER_OR_EQUAL)) {
          // 读取的文件块不为CORRUPT类型,并且文件块的时间大于最新的时间,表示非同一批的提交,退出处理
          break;
        }
        switch (r.getBlockType()) {
          case AVRO_DATA_BLOCK: // 数据块
            if (isNewInstantBlock(r) && !readBlocksLazily) { // 为新instant的block块并且为非延迟读取
              // 处理之前队列里的同一批block块
              processQueuedBlocksForInstant(currentInstantLogBlocks, scannedLogFiles.size());
            }
            // 将当前block块放入队列
            currentInstantLogBlocks.push(r);
            break;
          case DELETE_BLOCK: // 删除块
            if (isNewInstantBlock(r) && !readBlocksLazily) { // 新instant的block块并且为非延迟读取
              // 处理之前队列里的同一批block块
              processQueuedBlocksForInstant(currentInstantLogBlocks, scannedLogFiles.size());
            }
            // 将当前block块放入队列
            currentInstantLogBlocks.push(r);
            break;
          ...
  }

scan方法中,对于数据块和删除块的操作相同,即均会判断是不是新 instant对应的块(当前队列是否为空并且与上次写入队列的 block块的时间是否相同),若不是同一批,即instant的时间不同,那么调用 processQueuedBlocksForInstant开始处理,其核心代码如下

private void processQueuedBlocksForInstant(Deque<HoodieLogBlock> lastBlocks, int numLogFilesSeen) throws Exception {
    while (!lastBlocks.isEmpty()) { // 队列不为空
      // 取出队尾block
      HoodieLogBlock lastBlock = lastBlocks.pollLast();
      switch (lastBlock.getBlockType()) { // 判断block类型
        case AVRO_DATA_BLOCK: // 数据块
          // 处理数据块
          processAvroDataBlock((HoodieAvroDataBlock) lastBlock);
          break;
        case DELETE_BLOCK: // 删除块
          // 处理删除的所有key
          Arrays.stream(((HoodieDeleteBlock) lastBlock).getKeysToDelete()).forEach(this::processNextDeletedKey);
          break;
        case CORRUPT_BLOCK:
          LOG.warn("Found a corrupt block which was not rolled back");
          break;
        default:
          break;
      }
    }
  }

可以看到只要队列不为空,就会从队尾取出 Block块,然后根据类型处理数据块或者删除块。

2.1.1 处理数据块

使用 processAvroDataBlock对数据块进行处理,其核心代码如下

private void processAvroDataBlock(HoodieAvroDataBlock dataBlock) throws Exception {
    // 获取数据块的所有记录
    List<IndexedRecord> recs = dataBlock.getRecords();
    totalLogRecords.addAndGet(recs.size());
    for (IndexedRecord rec : recs) {
      // 构造HoodieRecord
      HoodieRecord extends HoodieRecordPayload> hoodieRecord =
          SpillableMapUtils.convertToHoodieRecordPayload((GenericRecord) rec, this.payloadClassFQN);
      // 处理HoodieRecord,其对应子类的具体实现
      processNextRecord(hoodieRecord);
    }
  }

首先获取该数据块的所有记录,然后调用具体子类的 processNextRecord进行处理,以 HoodieMergedLogRecordScanner的实现为例说明( HoodieUnMergedLogRecordScanner直接进行回调),其核心代码如下

protected void processNextRecord(HoodieRecord extends HoodieRecordPayload> hoodieRecord) throws IOException {
    // 获取key
    String key = hoodieRecord.getRecordKey();
    if (records.containsKey(key)) { // Map缓存中包含该key,该Map基于Disk实现
      // 将内容合并
      HoodieRecordPayload combinedValue = hoodieRecord.getData().preCombine(records.get(key).getData());
      // 放入Map缓存
      records.put(key, new HoodieRecord<>(new HoodieKey(key, hoodieRecord.getPartitionPath()), combinedValue));
    } else { // Map缓存中不包含该key
      // 直接放入缓存
      records.put(key, hoodieRecord);
    }
  }

可以看到首先会判断记录的key在缓存中是否存在,若存在,则将内容合并,否则放入缓存,该缓存基于磁盘实现,平衡内存占用,当无内存空间时,将数据写入磁盘。

2.1.2 处理删除块

处理删除块,会调用具体实现子类的 processNextDeletedKey方法来处理删除记录,其核心代码如下

protected void processNextDeletedKey(HoodieKey hoodieKey) {
    // 生成空的内容,然后放入缓存
    records.put(hoodieKey.getRecordKey(), SpillableMapUtils.generateEmptyPayload(hoodieKey.getRecordKey(),
        hoodieKey.getPartitionPath(), getPayloadClassFQN()));
  }

可以看到其会为对应的 key生成空的内容(达到软删除目的),然后放入缓存。

2.2 处理控制块

scan方法中处理控制块( command)的核心代码如下

case COMMAND_BLOCK:
            HoodieCommandBlock commandBlock = (HoodieCommandBlock) r;
            String targetInstantForCommandBlock =
                r.getLogBlockHeader().get(HoodieLogBlock.HeaderMetadataType.TARGET_INSTANT_TIME);
            switch (commandBlock.getType()) { // 控制类型
              case ROLLBACK_PREVIOUS_BLOCK: // 回滚
                int numBlocksRolledBack = 0;
                totalRollbacks.incrementAndGet();
                while (!currentInstantLogBlocks.isEmpty()) { // 当前队列不为空
                  // 取出块
                  HoodieLogBlock lastBlock = currentInstantLogBlocks.peek();
                  // 上个block为CORRUPT
                  if (lastBlock.getBlockType() == CORRUPT_BLOCK) {
                    // 弹出
                    currentInstantLogBlocks.pop();
                    numBlocksRolledBack++;
                  } else if (lastBlock.getBlockType() != CORRUPT_BLOCK
                      && targetInstantForCommandBlock.contentEquals(lastBlock.getLogBlockHeader().get(INSTANT_TIME))) { // 不为CORRUPT并且内容相等
                    // 弹出
                    currentInstantLogBlocks.pop();
                    numBlocksRolledBack++;
                  } else if (!targetInstantForCommandBlock
                      .contentEquals(currentInstantLogBlocks.peek().getLogBlockHeader().get(INSTANT_TIME))) { // 时间不相等
                    // 退出
                    break;
                  } 
                }
                break;
              default:
                throw new UnsupportedOperationException("Command type not yet supported.");
            }
            break;

可以看到,处理控制块时,会根据控制块的具体类型对已放入队列的块进行处理,现只支持回滚,对于队列中上一个类型为 CORRUPT的块或者类型不为 CORRUPT但时间相等的块,均从队列中弹出;当时间不相等时,则退出处理。

对于日志文件的 Block块的处理是基于 Deque处理,在处理完会放入基于磁盘的 Map中( HoodieMergedLogRecordScanner实现),然后可以通过 HoodieMergeLogRecordScanner#getRecords直接获取 Map或者通过 HoodieMergeLogRecordScanner#iterator获取该 Map的迭代器,现在社区对该 Map结构做重构,性能最多可提升6倍。

2.3 判断是否有下一个(hasNext)

在处理数据块之前,需要通过 HoodieLogFormatReader#hasNext方法判断是否还有下一个 HoodieLogBlock,其核心代码如下

public boolean hasNext() {
    if (currentReader == null) { // 为空则为false
      return false;
    } else if (currentReader.hasNext()) { // 有下一个
      return true;
    } else if (logFiles.size() > 0) { // 日志文件列表大于0
      try {
        HoodieLogFile nextLogFile = logFiles.remove(0);
        if (!readBlocksLazily) { // 非延迟读取block, 表示已经读完,则直接关闭
          this.currentReader.close();
        } else {
          // 加入集合中
          this.prevReadersInOpenState.add(currentReader);
        }
        // 生成新的reader
        this.currentReader =
            new HoodieLogFileReader(fs, nextLogFile, readerSchema, bufferSize, readBlocksLazily, false);
      } catch (IOException io) {
        throw new HoodieIOException("unable to initialize read with log file ", io);
      }
      // 判断当前reader有无下一个
      return this.currentReader.hasNext();
    }
    return false;
  }

可以看到,如果当前读取器( HoodieLogFileReader)为 null,那么表示已经读完所有日志文件,直接返回 false;否则若当前读取器有下一个,那么返回 true;否则若日志文件列表大小大于0,那么读取下一个日志文件,并生成新的读取器( HoodieLogFileReader),然后再判断是否有下一个;否则直接返回 false。其中, HoodieLogFileReader#hasNext方法通过读取魔数( #HUDI#)来判断是否有下一个,核心代码如下

private boolean hasNextMagic() throws IOException {
    long pos = inputStream.getPos();
    // 1. Read magic header from the start of the block
    inputStream.readFully(MAGIC_BUFFER, 0, 6);
    if (!Arrays.equals(MAGIC_BUFFER, HoodieLogFormat.MAGIC)) { // 比较是否为魔数
      return false;
    }
    return true;
  }

2.4 读取下一个(next)

在调用 hasNext后,若存在下一个,那么需要调用 next获取 HoodieLogBlock,获取 Block块的核心代码如下

private HoodieLogBlock readBlock() throws IOException {
    int blocksize = -1;
    int type = -1;
    HoodieLogBlockType blockType = null;
    Map<HeaderMetadataType, String> header = null;
    try {
      // 1 Read the total size of the block
      blocksize = (int) inputStream.readLong();
    } catch (EOFException | CorruptedLogFileException e) {
      // 异常时创建CORRUPT的Block
      return createCorruptBlock();
    }
    boolean isCorrupted = isBlockCorrupt(blocksize);
    if (isCorrupted) {
      return createCorruptBlock();
    }
    // 2. Read the version for this log format
    this.nextBlockVersion = readVersion();
    // 3. Read the block type for a log block
    if (nextBlockVersion.getVersion() != HoodieLogFormatVersion.DEFAULT_VERSION) {
      type = inputStream.readInt();
      Preconditions.checkArgument(type < HoodieLogBlockType.values().length, "Invalid block byte type found " + type);
      blockType = HoodieLogBlockType.values()[type];
    }
    // 4. Read the header for a log block, if present
    if (nextBlockVersion.hasHeader()) {
      header = HoodieLogBlock.getLogMetadata(inputStream);
    }
    int contentLength = blocksize;
    // 5. Read the content length for the content
    if (nextBlockVersion.getVersion() != HoodieLogFormatVersion.DEFAULT_VERSION) {
      contentLength = (int) inputStream.readLong();
    }
    // 6. Read the content or skip content based on IO vs Memory trade-off by client
    // TODO - have a max block size and reuse this buffer in the ByteBuffer
    // (hard to guess max block size for now)
    long contentPosition = inputStream.getPos();
    byte[] content = HoodieLogBlock.readOrSkipContent(inputStream, contentLength, readBlockLazily);
    // 7. Read footer if any
    Map<HeaderMetadataType, String> footer = null;
    if (nextBlockVersion.hasFooter()) {
      footer = HoodieLogBlock.getLogMetadata(inputStream);
    }
    // 8. Read log block length, if present. This acts as a reverse pointer when traversing a
    // log file in reverse
    long logBlockLength = 0;
    if (nextBlockVersion.hasLogBlockLength()) {
      logBlockLength = inputStream.readLong();
    }
    // 9. Read the log block end position in the log file
    long blockEndPos = inputStream.getPos();
    switch (blockType) {
      // 创建不同的Block
      case AVRO_DATA_BLOCK:
        if (nextBlockVersion.getVersion() == HoodieLogFormatVersion.DEFAULT_VERSION) {
          return HoodieAvroDataBlock.getBlock(content, readerSchema);
        } else {
          return HoodieAvroDataBlock.getBlock(logFile, inputStream, Option.ofNullable(content), readBlockLazily,
              contentPosition, contentLength, blockEndPos, readerSchema, header, footer);
        }
      case DELETE_BLOCK:
        return HoodieDeleteBlock.getBlock(logFile, inputStream, Option.ofNullable(content), readBlockLazily,
            contentPosition, contentLength, blockEndPos, header, footer);
      case COMMAND_BLOCK:
        return HoodieCommandBlock.getBlock(logFile, inputStream, Option.ofNullable(content), readBlockLazily,
            contentPosition, contentLength, blockEndPos, header, footer);
      default:
        throw new HoodieNotSupportedException("Unsupported Block " + blockType);
    }
  }

可以看到,对于 Block的读取,与写入时的顺序相同,然后根据不同类型生成不同的 Block

3. 总结

日志文件的读取,与日志文件写入的顺序相同。在读取后会将不同类型的 HoodieLogBlock先放入 Deque中处理,然后会根据不同的读取策略( HoodieUnMergedLogRecordScannerHoodieMergedLogRecordScanner)进行不同的处理,如 Merged策略会将同一key的内容进行合并(会处理删除和真实数据内容的合并),然后再将合并后的结果放入缓存中供读取;而 UnMerged策略则直接对 HoodieRecord进行回调处理。

相关实践学习
通过日志服务实现云资源OSS的安全审计
本实验介绍如何通过日志服务实现云资源OSS的安全审计。
目录
相关文章
|
22天前
|
监控 安全 搜索推荐
使用EventLog Analyzer进行日志取证分析
EventLog Analyzer助力企业通过集中采集、归档与分析系统日志及syslog,快速构建“数字犯罪现场”,精准追溯安全事件根源。其强大搜索功能可秒级定位入侵时间、人员与路径,生成合规与取证报表,确保日志安全防篡改,大幅提升调查效率,为执法提供有力证据支持。
|
6月前
|
存储 运维 监控
SelectDB 实现日志高效存储与实时分析,完成任务可领取积分、餐具套装/水杯/帆布包!
SelectDB 实现日志高效存储与实时分析,完成任务可领取积分、餐具套装/水杯/帆布包!
|
5月前
|
监控 容灾 算法
阿里云 SLS 多云日志接入最佳实践:链路、成本与高可用性优化
本文探讨了如何高效、经济且可靠地将海外应用与基础设施日志统一采集至阿里云日志服务(SLS),解决全球化业务扩展中的关键挑战。重点介绍了高性能日志采集Agent(iLogtail/LoongCollector)在海外场景的应用,推荐使用LoongCollector以获得更优的稳定性和网络容错能力。同时分析了多种网络接入方案,包括公网直连、全球加速优化、阿里云内网及专线/CEN/VPN接入等,并提供了成本优化策略和多目标发送配置指导,帮助企业构建稳定、低成本、高可用的全球日志系统。
632 54
|
3月前
|
监控 安全 NoSQL
【DevOps】Logstash详解:高效日志管理与分析工具
Logstash是ELK Stack核心组件之一,具备强大的日志收集、处理与转发能力。它支持多种数据来源,提供灵活的过滤、转换机制,并可通过插件扩展功能,广泛应用于系统日志分析、性能优化及安全合规等领域,是现代日志管理的关键工具。
435 0
|
5月前
|
自然语言处理 监控 安全
阿里云发布可观测MCP!支持自然语言查询和分析多模态日志
阿里云可观测官方发布了Observable MCP Server,提供了一系列访问阿里云可观测各产品的工具能力,包含阿里云日志服务SLS、阿里云应用实时监控服务ARMS等,支持用户通过自然语言形式查询
543 0
阿里云发布可观测MCP!支持自然语言查询和分析多模态日志
|
4月前
|
人工智能 运维 监控
Aipy实战:分析apache2日志中的网站攻击痕迹
Apache2日志系统灵活且信息全面,但安全分析、实时分析和合规性审计存在较高技术门槛。为降低难度,可借助AI工具如aipy高效分析日志,快速发现攻击痕迹并提供反制措施。通过结合AI与学习技术知识,新手运维人员能更轻松掌握复杂日志分析任务,提升工作效率与技能水平。
|
监控 Kubernetes Shell
使用阿里云日志服务采集查询kubernetes容器日志
使用阿里云日志收集查询kubernetes容器日志
3068 0
|
11月前
|
XML 安全 Java
【日志框架整合】Slf4j、Log4j、Log4j2、Logback配置模板
本文介绍了Java日志框架的基本概念和使用方法,重点讨论了SLF4J、Log4j、Logback和Log4j2之间的关系及其性能对比。SLF4J作为一个日志抽象层,允许开发者使用统一的日志接口,而Log4j、Logback和Log4j2则是具体的日志实现框架。Log4j2在性能上优于Logback,推荐在新项目中使用。文章还详细说明了如何在Spring Boot项目中配置Log4j2和Logback,以及如何使用Lombok简化日志记录。最后,提供了一些日志配置的最佳实践,包括滚动日志、统一日志格式和提高日志性能的方法。
3009 31
【日志框架整合】Slf4j、Log4j、Log4j2、Logback配置模板
|
10月前
|
监控 安全 Apache
什么是Apache日志?为什么Apache日志分析很重要?
Apache是全球广泛使用的Web服务器软件,支持超过30%的活跃网站。它通过接收和处理HTTP请求,与后端服务器通信,返回响应并记录日志,确保网页请求的快速准确处理。Apache日志分为访问日志和错误日志,对提升用户体验、保障安全及优化性能至关重要。EventLog Analyzer等工具可有效管理和分析这些日志,增强Web服务的安全性和可靠性。
264 9

热门文章

最新文章