一、概述 二、传输层 ServerAkka 中 创建 actor 需要继承 AbstractActor 类并重写它的初始行为方法 createReceive()，actor 接收消息后会触发 createReceive() 方法被调用，通过 receiveBuilder() 来接收消息以及它的类型，从而判断该如何处理消息。 123456public Receive createReceive() { return ReceiveBuilder.create() .match(RemoteHandshakeMessage.class, this::handleHandshakeMessage) .match(ControlMessages.class, this::handleControlMessage) .matchAny(this::handleMessage).build();} AkkaRpcActor 接收到的消息总共有3种。 握手消息 在客户端构造时会通过 ActorSelection 发送过来。收到消息后会检查接口、 ...

一、概述RpcService 是 RpcEndpoint 的运行时环境，是 Akka 中 ActorSystem 的封装。Flink 中 RpcService 也有多套， JobManager 和 TaskManager 进程中都有两套 RpcService。 二、结构设计2.1. 属性AkkaRpcService 是 RpcService 的唯一实现类。它除了持有 AkkaSystem (akkaSystem) 的引用外，还维护所有注册了的 RpcEndpoint的引用，为每个 RpcEndpoint 分配一个 ActerRef 并保存他们的对应关系 (actors)。 2.2. 方法RpcService 主要包含如下两个重要方法: 2.2.1. startServer()startServer() 方法用于启动 RpcEndpoint 中的 RpcServer。启动完成后，RpcEndpoint 中的 RpcServer 就能够对外提供服务了。 2.2.2. connect()connect(）方法用于连接远端 RpcEndpoint 并返回给调用方实现了 Rpc ...

一、概述Flink 内部组件之间的通信是用 Akka，比如 JobManager 和 TaskManager 之间的通信。而 Operator 之间的数据传输则利用 Netty 为不同的应用层通信协议 (RPC, FTP, HTTP 等) 提供支持。 1.1. Akka 概述Akka 是一套开源库，用于设计跨处理器和跨网络的可扩展弹性系统，借助 Akka 可以专注于满足业务需求，而不是编写底层代码来提供可靠的行为、容错和高性能。 引用本站文章 Akka 概述 Joker 二、Flink RPC 架构Flink 集群内部通信框架的最底层依赖于 Akka，定义了不同类型的消息，并且设计了通用的 RPC 通信组件，Flink 的所有提供 RPC 请求的集群组件(如 JobMaster、 TaskManager、 ResourceManager、 Di ...

一、概述

一、概述1.1. JVM 内存管理基于 Java 语言构建的应用，借助 JVM 提供的 GC 能力能够实现内存的自动管理，但会遇到一些基于 JVM 的内存管理问题。尤其对于大数据处理场景而言，需要处理非常庞大的数据，JVM 内存管理的问题就更加突出了，主要体现在以下几点： Java 对象存储密度相对较低 对于常用的数据类型，例如Boolean类型数据占16字节内存空间，其中对象头占字节，Boolean 属性仅占 1 字节，其余7字节做对齐填充。而实际上仅1字节就能够代表Boolean值，这种情况造成了比较严重的内存空间浪费 Full GC 影响系统性能使用 JVM 的垃圾回收机制对內存进行回收，在大数据量的情况下 GC 的性能会比较差，尤其对于大数据处理，有些数据对象处理完希望立即释放內存空间，但如果借助 JVM GC 自动回收，通常情况下会有秒级甚至分钟级别的延迟，这对系统的性能造成了非常大的影响 OutOfMemoryError 影响系统稳定性 系统出现对象大小分配超过 JVM 内存限制时，就会触发 OutofMemoryError。导致 JVM 宕机，影响整个数据处理进 ...

一、概述二、内存管理Flink 为了让用户更好的调整內存分配，达到资源的合理分配，在Flink1.10 31入了TaskManager 的内存管理，后续在 Flink1.11 版本引入了 JobManager 的内存管理，用户可以通过配置的方式合理的分配资源。不管是 TaskManager还是 JobManager都是单独的 JVM 进程，他们共用一套內存模型抽象 (TaskManager 的内存模型更加复杂), Flink 从一开始就选择了使用自主的内存管理，避开了JVM内存管理在大数据场景下的问题，提升了计算效率。 2.1. 概述Flink 从一开始就选择了使用自主的内存管理、避开了 JVM 内存管理在大数据场景下的问题，提升了计算效率。MemorySegment 就是 Flink 的内存抽象。默认情况下一个 MemorySegment 可以被看做是一个 32kb 大的内存块的抽象。NetworkBuffer, 是对 MemorySegment 的包装。Flink 在各个 TaskManager 之间传递数据时，使用的是这一层的抽象。最底内存抽象是 MemorySegmen ...

Flink 提供了三种状态后端(用于存储状态数据)，可以为所有 Flink 作业配置相同的状态后端，也可以为每个 Flink 作业配置指定的状态后端。状态可以存储在 Java 堆内存中或者堆外。 一、概述1.1. 什么是 state?Flink 在做计算的过程中经常需要存储中间状态，来避免数据丢失和状态恢复。选择的状态存储策略不同，会影响状态持久化如何和 checkpoint 交互, state 中存储着每条数据消费后数据的消费点（生产环境需要持久化这些状态），当 Job 因为某种错误或者其他原因导致重启时，就能够从 checkpoint 中的 state 数据进行恢复。 1.2. State Backends当需要对具体的某一种 State 做 Checkpoint 时，此时就需要具体的状态后端存储，Flink 内置提供了不同的状态后端存储，用于指定状态的存储方式和位置。状态可以存储在 Java 堆内存中或者堆外，在 Flink 安装路径下 conf 目录中的 flink-conf.yaml 配置文件中也有状态后端存储相关的配置。 Flink 支持基于每个 Job 单独设置状态后 ...

一、概述Spark Runtime Filter 在查询执行期间可以显著缩减中间数据量，并进而减少计算带来的成本。当查询执行中存在 Shuffle 阶段，由于 Executor 首先需要将 Shuffle 数据写到本地磁盈，然后由其他 Executor 通过网络拉取从磁密读取的 Shuffle 数据，所以其成本格外高昂!因此，能够缩减 Shuffle 的数据量一直以来是 Spark 性能优化的主要工作动向。Spark runtime filter 正是通过将 Join 一端具有高选择性的 Filter 推送到 Join 另一端来缩减 Shuffle 数据量的。 1.1. DPP&Runtime Filter 动态分区裁剪被裁剪的 join 一边必须是分区的,而且 join 另一边在 exchange 之前存在条件过滤，而且默认存在 broadcastJoin 的时候，才会进行分区裁剪Runtime Fitter 没有限制，但是Runtime Filter的适用条件更加严格 动态分区剪裁能够减少 source scan 的10，而 Runtime Fiter 不行，因为动态分 ...

一、概述Blockmanager 读取数据时，能从本地获取，则从 Diskstore 或 Memorystore 读取,没有就用使用块传输服务在不同节点间进行数据块传输～ 二、设计2.1. 传输服务客户端 BlockStoreClientBlockStoreClient 接口用于从 Executor 或外部服务读取 shuffle 文件和 RDD 块 2.1.1. BlockTransferServiceBlockTransferService 用于在不同节点之间传输数据块，NettyBlockTransferService 是一个基于 netty 实现的数据传输服务，NettyBlockTransferService 在 SparkEnv 初始化时创建、在 BlockManager 中初始化。 初始化 $init()$ NettyBlockTransferService 只有在其 $init()$ 方法被调用，即被初始化后才提供服务～ 上传数据块 $NettyBlockTransferService.uploadBlock()$ 方法利用 NettyBlockTran ...

一、概述存储内存池主要用于 RDD 的缓存，广播以及备份中。不像执行内存池需要维护每个 Task 的内存占用情况，存储内存池只提供了一个 _memoryUsed 的变量来进行当前内存的使用情况。 二、实现2.1. 结构2.1.1. 属性 _memoryUsed: 已经使用的内存大小 _memoryStore: 当前 StorageMemoryPool 所关联的 MemoryStore 三、内存管理3.1. 申请内存$acquireMemoryacquireMemory()$ 用于给 BlockId 对应的 Block 获取 numBytes 指定大小的内存。 首先计算要申请的内存大小 numBytes 与空闲空间 memoryFree 的差值 numBytesToFree，然后调用重载的 $acquireMemory()$ 方法申请获得内存。 1234def acquireMemory(blockId: BlockId, numBytes: Long): Boolean = lock.synchronized { val numBytesToFree = mat ...