正在总结中，等我😭～～～

一、概述Spark 提供了 CatalogManager，其内部通过一个 Map 类型的内存数据结构维护注册的 Catalog 实例。 二、实现2.1. 结构2.1.1. 属性 catalogs 保存 catalog 名字和 Class 的隐射关系 1private val catalogs = mutable.HashMap.empty[String, CatalogPlugin] 2.1.2. 方法 catalog() 用来查找特定名字的 Catalog，返回 CatalogPlugin 接口。 1234567def catalog(name: String): CatalogPlugin = synchronized { if (name.equalsIgnoreCase(SESSION_CATALOG_NAME)) { v2SessionCatalog } else { catalogs.getOrElseUpdate(name, Catalogs.load(name, conf)) }} ...

一、概述https://blog.csdn.net/onway_goahead/article/details/111186675 二、OrcFileFormat2.1. supportBatch()$supportBatch()$ 方法用来判断是否支持向量读取 123456override def supportBatch(sparkSession: SparkSession, schema: StructType): Boolean = { val conf = sparkSession.sessionState.conf conf.orcVectorizedReaderEnabled && schema.forall(s => OrcUtils.supportColumnarReads( s.dataType, sparkSession.sessionState.conf.orcVectorizedReaderNestedColumnEnabled)) //todo} 需要满足以下条件: 开启 spark.sql.or ...

对于每个行组，都会创建一个新的 VectorizedColumnReader 数组，每个请求的列都有一个 VectorizedColumnReader 实例。VectorizedParquetRecordReader 中的 $nextBatch()$ 方法会在 VectorizedColumnReader 数组中循环，将批次数据逐列读入相应的 ColumnVectors 中。

一、概述https://mp.weixin.qq.com/s/TVkuyBwCQmIfSR3VYFrGaQ https://www.learntospark.com/2020/02/how-to-read-and-write-parquet-in-apache-spark.html https://www.mianshigee.com/note/detail/221583thf/ https://www.sohu.com/a/507896675_120099884 https://mp.weixin.qq.com/s/HH8s6o7C50uaq2hs4axd0A 在初始化 VectorizedParquetRecordReader 并执行 $initBatch()$ 方法创建 ColumnBatch 实例和相关 ColumnVector 实例后，就可以通过 $nextKeyValue()$ 方法逐批读取 parquet 文件。 在 SparkSQL 优化中，会把查询的过滤条件，下推到靠近存储层，这样可以减少从存储层读取的数据量。其次在真正读取过滤数据时，Spark并不自己 ...

根据数据编码方法，VectorizedColumnReader 会创建相应的 VectorizedValuesReader 实现实例来读取和解码数据。 目前，VectorizedValuesReader 有两个版本: VectorizedRleValuesReader(用于 RLE/Bit-PackingHybrid 编码) 和 VectorizedPlainValuesReader(用于普通编码)。 目录解码逻辑封装在 VectorizedColumnReader 的 $decodeDictionaryIds()$ 方法中，页面数据的读取和解码以指定的批次大小批量进行。批次中读取和解码的列数据将写入 ColumnVector 并返回 VectorizedParquetRecordReader，然后由上游调用者与 ColumnBatch 中的其他 ColumnVector 一起使用。 https://dataninjago.com/category/deep-dive-parquet-for-spark/ https://www.waitingforcode.com/apach ...

一、概述https://mp.weixin.qq.com/s/TVkuyBwCQmIfSR3VYFrGaQ https://www.learntospark.com/2020/02/how-to-read-and-write-parquet-in-apache-spark.html https://www.mianshigee.com/note/detail/221583thf/ https://www.sohu.com/a/507896675_120099884 https://mp.weixin.qq.com/s/HH8s6o7C50uaq2hs4axd0A 1.1. Parquet二、ParquetFileFormat 2.1. supportBatch()1234override def supportBatch(sparkSession: SparkSession, schema: StructType): Boolean = { val conf = sparkSession.sessionState.conf ParquetUtils.isBa ...

一、概述1234567val df: DataFrame = spark.read.format("jdbc") .option("url", "jdbc:mysql://127.0.0.1:3306/siros?...") .option("driver", "com.mysql.cj.jdbc.Driver") .option("user", "root") .option("password", "3494269") .option("dbtable", "satcat") .load() 二、read()创建 DataFrameReader 对象，进行数据读取任务。 1def read: DataFrameReader = new DataFrameReader(self) DataFrameReader 用于从外部数据源加载数据到 ...

正在总结中，等我😭～～～

一、概述MemoryManager 是 Flink 中管理托管内存的组件，其管理的托管内存只使用堆外内存。在批处理中用在排序、Hash 表和中间结果的缓存中，在流计算中作为 RocksDBStateBackend 的内存。在 1.10 之前的 Flink 版本中，MemoryManager 负责 TaskManager 的所有内存，1.10 版本中，MemoryManager 的管理范围缩小为 Slot 级别，即为 Task 管理内容， TaskManager 为每个 Slot 分配相同的内容，Task 不能使用超过其 Slot 分配的资源。这样的实现并不完美，但是相比 1.10 之前的版本，能够更好地隔离任务，系统更加稳定。 需要注意的是，并不是所有的算子都会使用 MemoryManager 申请内存空间，这个步骤主要针对批计算类型的算子，例如 HashJoinOperator SorMergeJoinOperator 和 SortOperator等，这些算子往往需要借助非常大的内存空间进行数据的排序等操作。MemoryManager 主要通过内部接口 MemoryPo ...