Netty 入门学习

2025-01-09

前言

学习Spark源码绕不开通信，Spark通信是基于Netty实现的，所以先简单学习总结一下Netty。

Spark 通信历史

最开始: Akka
Spark 1.3：开始引入Netty，为了解决大块数据（如Shuffle）的传输问题
Spark 1.6：支持配置使用 Akka 或者 Netty
Spark 2：完全废弃Akka,全部使用Netty

Akka 是一个用 Scala 编写的库，用于简化编写容错的、高可伸缩性的 Java 和 Scala 的 Actor 模型应用。
Spark 借鉴Akka 通过 Netty 实现了类似的简约版的Actor 模型

more >>

展开全文 >>

log4j 单独设置某个类或者某个包的级别

2024-12-27

原本的设置

log4j.rootLogger = INFO, dailyfile

log4j.appender.dailyfile.DatePattern=.yyyy-MM-dd
log4j.appender.dailyfile.Threshold = INFO
log4j.appender.dailyfile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.dailyfile.File = /var/log/zeppelin/zeppelin.log
log4j.appender.dailyfile.layout = org.apache.log4j.PatternLayout
log4j.appender.dailyfile.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1} %l: %m%n

配置解析

rootLogger

rootLogger：根Logger (日志写出器)，可以理解为默认日志写出器

语法为： log4j.rootLogger = [ level ] , appenderName, …
[level]：根Logger的写出日志的级别，这里的level=INFO
appenderName：Appenders组件名，这里的appenderName=dailyfile，名称可以自定义，可以同时写多个，逗号分隔

more >>

展开全文 >>

Hudi 源码 | 索引总结 - tag/tagLocation

2024-07-15

前言

接上篇文章和之前的总结的源码文章，本文总结源码 tag/tagLocation ，对应功能：根据索引信息判断记录是否存在，如果不存在，代表是新增数据，如果记录存在则代表是更新数据，需要找到并设置 currentLocation。

tag

AbstractWriteHelper.tag

private I tag(
    I dedupedRecords, HoodieEngineContext context, HoodieTable<T, I, K, O> table) {
  // perform index loop up to get existing location of records
  // 执行索引循环以获取记录的现有位置
  // 对于 Java Client 这里 table 为 HoodieJavaCopyOnWriteTable
  return table.getIndex().tagLocation(dedupedRecords, context, table);
}

table.getIndex()

对于 Java Client 这里 table 为 HoodieJavaCopyOnWriteTable ， HoodieJavaCopyOnWriteTable.getIndex() -> HoodieJavaTable.getIndex

protected HoodieIndex<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> getIndex(HoodieWriteConfig config, HoodieEngineContext context) {
  return JavaHoodieIndex.createIndex(config);
}

public static HoodieIndex createIndex(HoodieWriteConfig config) {
  // first use index class config to create index.
  if (!StringUtils.isNullOrEmpty(config.getIndexClass())) {
    Object instance = ReflectionUtils.loadClass(config.getIndexClass(), config);
    if (!(instance instanceof HoodieIndex)) {
      throw new HoodieIndexException(config.getIndexClass() + " is not a subclass of HoodieIndex");
    }
    return (JavaHoodieIndex) instance;
  }

  // TODO more indexes to be added
  // 从这里看出，当前版本（0.9.0），Java Client 只支持两种索引类型：INMEMORY 和 BLOOM
  switch (config.getIndexType()) {
    case INMEMORY:
      return new JavaInMemoryHashIndex(config);
    case BLOOM:
      return new JavaHoodieBloomIndex(config);
    default:
      throw new HoodieIndexException("Unsupported index type " + config.getIndexType());
  }
}

因为指定了索引类型为 BLOOM , 所以这里返回 JavaHoodieBloomIndex 。

more >>

展开全文 >>

Hudi 索引总结 - Parquet布隆过滤器写入过程

2024-07-10

前言

上篇文章提到：索引的逻辑主要是根据 parquet 文件中保存的索引信息，判断记录是否存在，如果不存在，代表是新增数据，如果记录存在则代表是更新数据，需要找到并设置 currentLocation。对于布隆索引来说，这里的索引信息其实是布隆过滤器，本篇文章主要是先总结布隆过滤器是如何保存到 parquet 文件中的（主要是源码调用逻辑）。

我们新写一个表的时候，最开始是没有parquet文件的，所以顺序应该是先将布隆过滤器写到 parquet 文件中，下次写数据的时候，先读取 parquet 文件中的布隆过滤器来验证表中是否存在该记录，有助于实现高效的更新和删除操作。