【Flink网络数据传输(3)】RecordWriter的能力:实现数据分发策略或广播到下游InputChannel

文章目录

  • 一.创建RecordWriter实例都做了啥
    • 1. 根据recordWrites数量创建不同的代理类
    • 2. 创建RecordWriters
    • 3. 单个RecordWriter的创建细节
  • 二. RecordWriter包含的主要组件
    • 1. RecordWriter两种实现类分别实现分发策略和广播
    • 2. ChannelSelectorRecordWriter的发送策略
      • 2.1. ChannelSelector根据实现类实现不同发送策略
    • 2. BroadcastRecordWriter广播所有元素到下游InputChannel

StreamTask节点中的中间结果数据元素最终通过RecordWriterOutput实现了网络输出,RecordWriterOutput底层依赖RecordWriter组件完成数据输出操作,接下来我们深入了解RecordWriter的设计和实现。

一.创建RecordWriter实例都做了啥

在StreamTask构造器方法中会直接创建RecordWriter实例,用于输出当前任务产生的Intermediate Result数据

1. 根据recordWrites数量创建不同的代理类

其中createRecordWriterDelegate根据recordWrites的数量创建对应的RecordWriterDelegate代理类。

protected StreamTask(  Environment environment,  @Nullable TimerService timerService,  Thread.UncaughtExceptionHandler uncaughtExceptionHandler,  StreamTaskActionExecutor actionExecutor,  TaskMailbox mailbox)  throws Exception {...this.recordWriter = createRecordWriterDelegate(configuration, environment);...}
//1. 如果recordWrites数量等于1,则创建SingleRecordWriter代理类;
//2. 如果recordWrites数量等于0,则创建NonRecordWriter代理类;
//3. 其他情况则创建MultipleRecordWriters代理类。@VisibleForTesting  
public static <OUT>  RecordWriterDelegate<SerializationDelegate<StreamRecord<OUT>>>  createRecordWriterDelegate(  StreamConfig configuration, Environment environment) {  List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWrites =  createRecordWriters(configuration, environment);  if (recordWrites.size() == 1) {  return new SingleRecordWriter<>(recordWrites.get(0));  } else if (recordWrites.size() == 0) {  return new NonRecordWriter<>();  } else {  return new MultipleRecordWriters<>(recordWrites);  }  
}

 

2. 创建RecordWriters

获取StreamTask的所有输出边放到RecordWriters中,返回创建的RecordWriter集合。

private static <OUT> List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters(StreamConfig configuration,Environment environment) {// 创建RecordWriter集合List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWriters = new ArrayList<>();// 获取输出的StreamEdge(所有的输出边)List<StreamEdge> outEdgesInOrder = configuration.getOutEdgesInOrder(environment.getUserClassLoader());// 获取chainedConfigs参数Map<Integer, StreamConfig> chainedConfigs = configuration.getTransitiveChainedTaskConfigsWithSelf(environment.getUserClassLoader());// 遍历输出节点,分别创建RecordWriter实例for (int i = 0; i < outEdgesInOrder.size(); i++) {StreamEdge edge = outEdgesInOrder.get(i);recordWriters.add(createRecordWriter(edge,i,environment,environment.getTaskInfo().getTaskName(),chainedConfigs.get(edge.getSourceId()).getBufferTimeout()));}return recordWriters;
}

 

3. 单个RecordWriter的创建细节

看单个RecordWriter的创建过程,包括分区策略缓存结果的ResultPartition、最后通过RecordWriterBuilder创建RecordWriter、以及设定MetricGroup监控RecordWriter指标并输出。

  1. 创建outputPartitioner:StreamPartitioner分区策略会被应用在RecordWriter中,例如DataStream.rebalance()操作就会创建RebalancePartitioner作为StreamPartitioner的实现类,并通过RebalancePartitioner选择下游InputChannel,实现数据元素按照指定的分区策略下发
  2. ResultPartition: ResultPartition内部会在本地存储需要下发的Buffer数据,并等待下游节点向上游节点发送数据消费请求
  3. 通过RecordWriterBuilder创建RecordWriter,在创建过程中会设定outputPartitioner、bufferTimeout以及bufferWriter等参数。
  4. 为RecordWriter设定MetricGroup,用于监控指标的采集和输出。
private static <OUT> RecordWriter<SerializationDelegate<StreamRecord<OUT>>> createRecordWriter(StreamEdge edge,int outputIndex,Environment environment,String taskName,long bufferTimeout) {@SuppressWarnings("unchecked")// 获取边上的StreamPartitionerStreamPartitioner<OUT> outputPartitioner = (StreamPartitioner<OUT>) edge.getPartitioner();LOG.debug("Using partitioner {} for output {} of task {}", outputPartitioner, outputIndex, taskName);// 获取ResultPartitionWriterResultPartitionWriter bufferWriter = environment.getWriter(outputIndex);// 初始化Partitionerif (outputPartitioner instanceof ConfigurableStreamPartitioner) {int numKeyGroups = bufferWriter.getNumTargetKeyGroups();if (0 < numKeyGroups) {((ConfigurableStreamPartitioner) outputPartitioner).configure(numKeyGroups);}}// 创建RecordWriterRecordWriter<SerializationDelegate<StreamRecord<OUT>>> output = new RecordWriterBuilder<SerializationDelegate<StreamRecord<OUT>>>().setChannelSelector(outputPartitioner).setTimeout(bufferTimeout).setTaskName(taskName).build(bufferWriter);// 设定MetricGroup监控output.setMetricGroup(environment.getMetricGroup().getIOMetricGroup());return output;
}

 

二. RecordWriter包含的主要组件

RecordWriter内部主要包含RecordSerializer和ResultPartitionWriter两个组件。

  • RecordSerializer用于对输出到网络中的数据进行序列化操作,将数据元素序列化成Bytes[]二进制格式,维护Bytes[]数据中的startBuffer及position等信息
  • ResultPartitionWriter是ResultPartition实现的接口,提供了将数据元素写入ResultPartiton的方法,例如addBufferConsumer()方法就是将RecordSerializer序列化的BufferConsumer数据对象添加到ResultPartition队列并进行缓存,供下游InputGate消费BufferConsumer对象

 

1. RecordWriter两种实现类分别实现分发策略和广播

RecordWriter主要有两种实现类:ChannelSelectorRecordWriter和BroadcastRecordWriter。

  • ChannelSelectorRecordWriter根据ChannelSelector选择下游节点的InputChannel,ChannelSelector内部基于StreamPartitoner获取不同的数据下发策略,最终实现数据重分区。
  • BroadcastRecordWriter对应广播式数据下发,即数据元素会被发送到下游所有的InputChannel中。当用户执行了Broadcast操作时,就会创建BroadcastRecordWriter实现数据元素的广播下发操作。

如代码:通过RecordWriterBuilder创建RecordWriter,此时会根据selector.isBroadcast()条件选择创建ChannelSelectorRecordWriter还是BroadcastRecordWriter实例。

public RecordWriter<T> build(ResultPartitionWriter writer) {if (selector.isBroadcast()) {return new BroadcastRecordWriter<>(writer, timeout, taskName);} else {return new ChannelSelectorRecordWriter<>(writer, selector, timeout, taskName);}
}

 

2. ChannelSelectorRecordWriter的发送策略

ChannelSelectorRecordWriter控制数据元素发送到下游的哪些InputChannel中。如代码,调用channelSelector.selectChannel(record)选择下游的InputChannel。

对于非广播类型的分区器,最终都会创建ChannelSelectorRecordWriter实现StreamRecord数据的下发操作。

public void emit(T record) throws IOException, InterruptedException {emit(record, channelSelector.selectChannel(record));
}

 

2.1. ChannelSelector根据实现类实现不同发送策略

ChannelSelector的实现类主要有StreamPartitioner、RoundRobinChannelSelector和OutputEmitter三种。

  • StreamPartitioner:DataStream API中物理操作指定的分区器,例如当用户调用DataStream.rebalance()方法时,会创建RebalencePartitioner。在StreamTask执行的过程中,会获取相应的StreamPartitioner应用在ChannelSelectorRecordWriter中,实现对数据元素分区的选择。
  • RoundRobinChannelSelector:ChannelSelector的默认实现类,提供了对Round-Robin策略的支持,以轮询的方式随机选择一个分区输出数据元素。
  • OutputEmitter:适用于BatchTask,须配合ShipStrategyType使用,通过ShipStrategyType执行的策略输出数据。

 

2. BroadcastRecordWriter广播所有元素到下游InputChannel

BroadcastRecordWriter的实现就比较简单了,在BroadcastRecordWriter中不需要ChannelSelector组件选择数据元素分区,直接将所有的数据元素广播发送到下游所有InputChannel中即可。

在这里插入图片描述

 

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/726392.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

ArmSoM规划开发基于RK3576的开发套件

ArmSoM正计划推出一款新的产品&#xff0c;这款产品将采用强大的RK3576芯片。 本文将为您介绍我们的新产品搭载的RK3576性能参数&#xff0c;以及它如何为您提供卓越的性能和功能。 RK3576处理器 RK3576处理器是一款强大的处理器&#xff0c;具备出色的性能和多样化的功能&a…

8、Linux-软件安装:rpm和yum;配置yum阿里云镜像源

一、介绍 Linux安装软件有两种方式&#xff0c; ①rpm&#xff1a;安装已有的安装包&#xff0c;类似于Windows中双击exe的安装包程序 ②yum&#xff1a;拉取远程仓库的文件&#xff0c;类似于python的pip install 区别&#xff1a;假设软件A依赖软件B&#xff0c;软件B依赖…

速盾网络:cdn加速技术和云计算的区别

CDN加速技术和云计算是两种不同的技术&#xff0c;虽然都与网络性能和可扩展性有关&#xff0c;但它们在功能和应用方面存在一些不同之处。 首先&#xff0c;CDN加速技术&#xff08;Content Delivery Network&#xff09;是一种通过将网站内容分布到全球各地的服务器上&#…

【C++庖丁解牛】C++内存管理 | new和delete的使用以及使用原理

&#x1f4d9; 作者简介 &#xff1a;RO-BERRY &#x1f4d7; 学习方向&#xff1a;致力于C、C、数据结构、TCP/IP、数据库等等一系列知识 &#x1f4d2; 日后方向 : 偏向于CPP开发以及大数据方向&#xff0c;欢迎各位关注&#xff0c;谢谢各位的支持 目录 1. C/C内存分布2. C语…

【C语言】走迷宫之推箱子

前言&#xff1a; 在上一篇文章当中我介绍了一个走迷宫的写法&#xff0c;但是那个迷宫没什么可玩性和趣味性&#xff0c;所以我打算在迷宫的基础上加上一个推箱子&#xff0c;使之有更好的操作空间&#xff0c;从而增强了游戏的可玩性和趣味性。 1. 打印菜单 void menu() {…

Day12:信息打点-Web应用源码泄漏开源闭源指纹识别GITSVNDS备份

目录 开源-CMS指纹识别源码获取方式 闭源-习惯&配置&特性等获取方式 闭源-托管资产平台资源搜索监控 思维导图 章节点 Web&#xff1a;语言/CMS/中间件/数据库/系统/WAF等 系统&#xff1a;操作系统/端口服务/网络环境/防火墙等 应用&#xff1a;APP对象/API接口/微…

ArmSoM Rockchip系列产品 通用教程 之 Ethernet 使用

1. Ethernet 简介​ ArmSoM系列产品主要使用到了两款以太网芯片 RTL8211F&#xff0d;CGRTL8125B 2. RTL8211F&#xff0d;CG芯片​ 在ArmSoM系列产品中&#xff0c;ArmSoM-Aim7使用的是RTL8211F&#xff0d;CG千兆以太网芯片 2.1 驱动​ drivers/net/ethernet/stmicro/s…

Spring MVC RequestParamMethodArgumentResolver原理解析

在Spring MVC框架中&#xff0c;RequestParamMethodArgumentResolver是一个核心的类&#xff0c;它负责解析HTTP请求中的参数&#xff0c;并将其绑定到处理器方法的参数上。这对于从请求中获取数据并将其传递给后端业务逻辑是至关重要的。本文将详细解析RequestParamMethodArgu…

Flink JobGraph构建过程

文章目录 前言JobGraph创建的过程总结 前言 在StreamGraph构建过程中分析了StreamGraph的构建过程&#xff0c;在StreamGraph构建完毕之后会对StreamGraph进行优化构建JobGraph&#xff0c;然后再提交JobGraph。优化过程中&#xff0c;Flink会尝试将尽可能多的StreamNode聚合在…

Vue 导出前端数据报表为xlsx文件

文章目录 前言一、添加依赖包二、新建导出功能按钮组件三、使用示例前言 导出数据报表基本上有两种形式,第一种是前端请求数据之后,后端将数据导出为文件,前端再将文件下载下来;第二种是前端请求数据之后,前端将数据保存到文件。 本文将讲解第二种方法,前端请求数据之后…

Sharding-JDBC源码解析与vivo的定制开发

作者&#xff1a;vivo IT 平台团队 - Xiong Huanxin Sharding-JDBC是在JDBC层提供服务的数据库中间件&#xff0c;在分库分表场景具有广泛应用。本文对Sharding-JDBC的解析、路由、改写、执行、归并五大核心引擎进行了源码解析&#xff0c;并结合业务实践经验&#xff0c;总结…

vue3之Prop特性注意点

1、Ts中接收父组件传递参数prop的定义写法&#xff1a; <script setup lang"ts">defineProps<{title?: stringlikes?: number}>() </script>2、所有的 props 都遵循着单向绑定原则&#xff0c;props 因父组件的更新而变化&#xff0c;子组件中不…

基于 Vue3 学习状态管理器:pinia

pinia 基本概念 Pinia 是 Vue 的存储库&#xff0c;Pinia和Vuex一样都是是vue的全局状态管理器&#xff0c;它允许跨组件/页面共享状态。实际上&#xff0c;其实Pinia就是Vuex5&#xff0c;官网也说过&#xff0c;为了尊重原作者&#xff0c;所以取名 pinia&#xff0c;而没有…

raylib库在CodeBlocks上的配置

raylib下载 raylib | A simple and easy-to-use library to enjoy videogames programming CodeBlocks

监控易对多云平台的运维管理方案

随着企业信息化建设的不断深入&#xff0c;越来越多的企业开始采用多云策略&#xff0c;以充分利用不同云服务提供商的优势&#xff0c;实现业务的高可用性、灵活性和成本效益。然而&#xff0c;多云环境也带来了运维管理的复杂性&#xff0c;如何有效监控和管理多个云平台的资…

本地部署websocket服务端并结合内网穿透实现固定公网地址连接

文章目录 1. Java 服务端demo环境2. 在pom文件引入第三包封装的netty框架maven坐标3. 创建服务端,以接口模式调用,方便外部调用4. 启动服务,出现以下信息表示启动成功,暴露端口默认99995. 创建隧道映射内网端口6. 查看状态->在线隧道,复制所创建隧道的公网地址加端口号7. 以…

CGAL 5.6.1 - Algebraic Foundations

1. 引言 CGAL 的目标是精确计算非线性对象&#xff0c;特别是定义在代数曲线和曲面上的对象。因此&#xff0c;表示多项式、代数扩展和有限域的类型在相关的实现中扮演着更加重要的角色。为了跟上这些变化&#xff0c;我们引入了这个软件包。由于引入的框架必须特别支持多项式…

【杂言】迟到的 2024 展望

研一下开学已有半月&#xff0c;本来想在家写的新年展望拖到了现在。翻看 2021 年的展望&#xff0c;我发现 flag 基本达成了&#xff08;除了 12 点睡觉&#xff09;&#xff0c;所以给新的一年立下大方向也是很有必要的。也许等到 60 岁我再回看&#xff0c;也是一件趣事吧。…

docker常见命令

命令 说明 docker pull 拉取镜像 docker push 推送镜像到DockerRegistry docker images 查看本地镜像 docker rmi 删除本地镜像 docker run 创建并运行容器&#xff08;不能重复创建&#xff09; docker stop 停止指定容器 docker start 启动指定容器 docker r…