一、概述HDFS 除了定义 RPC 调用接口外，还定义了流式接口，流式接口是 HDFS 中基于 TCP 或者 HTTP 实现的接口。在 HDFS 中，流式接口包括了基于 TCP 的 DataTransferProtocol 接口，以及 HA 架构中 Active Namenode 和 Standby Namenode 之间的 HTTP 接口。 二、实现

HDFS 添加了Namenode 高可用性支持，但是所有的 HA 实现方案都依赖于一个保存 editlog 的共享存储。这个共享存储必须是高可用的，并且能够被集群中的所有 Namenode 同时访问。 一、概述在 Quorum Journal 模式之前，HDFS 中使用最多的共享存储方案是 NAS+NFS。但是这种方案有个缺点，就是为了预防脑裂的情况，它要求有一个互斥脚本在 Namenode 发生故障切换时关闭上一个活动节点，或者阻止上一个活动节点访问共享存储。为了解决这个问题，cloudera 提供了 Quorum Journal 设计方案，这是一个基于 Paxos 算法实现的 HA 方案 二、实现2.1. 相关类 JournalNode 运行在 N 台独立的物理机器上，它将 editlog 文件保存在 JournalNode 的本地磁盘上，同时 JouralNode 还对外提供 RPC 接口 QJournalProtocol 以执行远程读写 editlog 文件的功能。 QuorumJournalManager 运行在 Namenode 上（目前 HA 集群中只有两个 Na ...

一、概述为了简化应用程序提交的复杂性，Spark提供了各种应用程序提交的统一入口，即 spark-submit 脚本，应用程序的提交都间接或直接地调用了该脚本。 二、源码～Spark 任务的第一步: 如何提交用户编写的程序🤔️～ 2.1. /bin/spark-submitSpark 提供了各种应用程序 application 提交的统一入口, 即 spark-submit 脚本, 应用程序的提交都间接或直接地调用了该脚本 https://blog.csdn.net/qq_26838315/article/details/115175653 用户应用程序可以使用 bin/spark-submit 脚本来启动。spark-submit 脚本负责使用 Spark 及其依赖关系设置类路径，并可支持 Spark 支持的不同群集管理器和部署模式。 1exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@" 所有其他脚本最终都调 ...

一、概述 二、实现https://mp.weixin.qq.com/s/Z-VOIkb-ubGlyxZxOMcYDA 2.1. 属性 minMemoryMapBytes: 使用内存映射(memory map)读取文件的最小阈值 由配置项 spark.storage.memoryMapThreshold 指定，默认值2M。当磁盘中的文件大小超过该值时，就不会直接读取，而用内存映射文件来读取，提高效率。 maxMemoryMapBytes: 使用内存映射读取文件的最大阈值 由配置项 spark.storage.memoryMapLimitForTests 指定。它是个测试参数，默认值为不限制。 blockSizes: 维护块ID与其对应大小之间的映射关系的ConcurrentHashMap。 三、BlockData3.1. DiskBlockData3.1.1. 方法 $toChunkedByteBuffer()$ 在数据量比较大的时候，因为每次申请的内存块大小有限制 maxMemoryMapBytes，因此须要切分红多个块 $toChunkedByteBuffer()$ 方 ...

一、理论Spark 本身是基于内存计算的架构，数据的存储也主要分为内存和磁盘两个路径。在 Spark 当中，磁盘都用在哪些地方呢🤔️～ 溢出临时文件 存储 Shuffle 中间文件 缓存分布式数据集 磁盘的第三个作用就是缓存分布式数据集。也就是说，凡是带DISK字样的存储模式，都会把内存中放不下的数据缓存到磁盘 Spark 将经常被重要的数据缓存到内存中以提升数据读取速度，当内存容量有限时，则将数据存入磁盘中或根据最近最少使用页面置换算法将内存中使用频率较低的文件空间收回，从而让新的数据进来。Spark 则根据存储位置、是否可序列化和副本数目这凡个要素将数据存储分为多种存储级别。 二、实现2.1.架构设计2.1.1. DiskBlockManagerDiskStore 负责 Spark 磁盘存储。依赖于 DiskBlockManager, DiskBlockManager 负责为逻辑的 Block 与数据写入磁盘的位置之间建立逻辑的映射关系 引用本站文章 ...

一、概述DataSourceV2 是 Spark 2.x 新推出的 API，用来和外部数据存储进行集成。原来的 SessionCatalog 也暴露出弊端和不足，缺少对表的元数据进行操作，比如创建、修改、删除表等，为了适应新的数据源特性，Spark 推出了新的接口: CatalogPlugin。 引用站外地址，不保证站点的可用性和安全性 Spark Catalog深入理解与实战 Joker 引用站外地址，不保证站点的可用性和安全性 Spark 笔记之 Catalog Joker 二、设计2.1. 元数据模型Spark 的元数据模型定义了任务的元数据结构，如数据库、表、视图、函数等，Spark 定义了 4 类接口分别对应于 4 种元数据类型，元数据类型之间的层次 ...

一、概述Spark 中的 DataSet 和 Dateframe API 支持结构化分析。结构化分析的一个重要的方面是管理元数据。这些元数据可能是一些临时元数据(比如临时表）SQLContext 上注册的 UDF 以及持久化的元数据。 Spark 的早期版本并没有标准的 API 來访问这些元数据的。用户通常使用查询语句(比如 show tables) 来查询这些元数据。这些查询通常需要操作原始的字符串，而且不同元数据类型的操作也是不一样的。Spark 2.0 中添加了标准的 API(称为 Catalog) 来访问 Spark SQL 中的元数据。 引用站外地址，不保证站点的可用性和安全性 一篇文章了解 Spark 3.x 的 Catalog 体系 Joker 二、设计2.1. Spark 2.x 的 Catalog 体系2.1.1. 设计 在 Spark 2.× 中，Spark SQL 中的 Catalog 体系实现以 ...

一、概述Spark SQL command 主要指的是一些 non-query 逻辑计划。 二、设计 2.1. AnalysisOnlyCommand仅解析(analyzed) 子节点，不优化 (optimized) 子节点的命令。 HandleSpecialCommand 处理完成 analysis 时需要通知的特殊命令的规则。 HandleSpecialCommand 规则应在所有其他分析规则运行后运行。 12345678object HandleSpecialCommand extends Rule[LogicalPlan] { override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsWithPruning(_.containsPattern(COMMAND)) { case c: AnalysisOnlyCommand if c.resolved => checkAnalysis(c) c.markAsAnalyzed(A ...

一、概述Spark SQL 最终将 SQL 语句经过逻辑算子树转换成物理算子树。在物理算子树中，叶子类型的 SparkPlan 节点负责 “从无到有” 地创建 RDD，每个非叶子类型的 SparkPlan 节点等价于在RDD 上进行一次 Transformation，即通过调用 execute() 函数转换成新的 RDD，最终执行 collect() 操作触发计算，返回结果给用户。SparkPlan 在对 RDD 做 Transformation 的过程中除对数据进行操作外，还可能对 RDD 的分区做调整。此外，SparkPlan 除实现 execute 方法外，还有一种情况是直接执行 executeBroadcast 方法，将数据广播到集群上。 二、模块具体来看，SparkPlan 的主要功能可以划分为 3 大块。首先，每个 SparkPlan 会记录其元数据(Metadata) 与指标(Metric)信息，这些信息以 Key-Value 的形式保存在 Map 数据结构中，统称为 SparkPlan 的 Metadata 与 Metric 体系。其次，在对 RDD 进行 Tran ...

一、概述AppStatusStore 用来存储 Application 的状态数据，Spark Web UI 及 REST API 需要的数据都取自它。 二、设计AppStatusStore 的构造依赖于两个要素: 一为键值对存储 KVStore，二为 App 状态监听器 AppStatusListener 2.1. KVStoreInMemoryStore 是在内存中维护的键值对存储；LevelDB 则是借助 Google 开源的 KV 数据库来实现，可以持久化到磁盘。ElementTrackingStore 额外加上了跟踪元素个数的功能，可以根据元素个数阈值触发特定的操作，但它更多地是个包装类，需要依赖于 InMemoryStore 或者 LevelDB。 2.1.1. ElementTrackingStoreElementTrackingStore 的初始化依赖于 InMemoryStore，它的多数方法都是直接代理了 InMemoryStore 的方法。为了实现跟踪元素数并触发操作的功能，其内部维护了一个类型与触发器(通过内部样例类 Trigger 定义)的映射关系， ...