辽宁省电力建设网站/pc端百度
背景
本文基于 Spark 3.3, 最近在用 Spark Api 做 DataSet[Row]
转换的时候遇到了一些 Spark内部转换的问题, 在此记录一下。
杂谈
我们知道在Spark中分为 InternalRow
和Row
, 前者是 Spark 内部的使用的一行数据的表示,后者是给Spark开发者使用的行数据表示。
在Spark中如果涉及到 InternalRow
和Row
转换的时候,这个时候就会用到 ExpressionEncoder[Row]
来做序列化和反序列化,而获取 ExpressionEncoder[Row]
的方式一般就是调用RowEncoder.apply(StructType)
方法。比如在delta 1.0.0
用到的 Row转换:
val joinedRowEncoder = RowEncoder(joinedPlan.schema)val outputRowEncoder = RowEncoder(deltaTxn.metadata.schema).resolveAndBind()val processor = new JoinedRowProcessor(targetRowHasNoMatch = resolveOnJoinedPlan(Seq(col(SOURCE_ROW_PRESENT_COL).isNull.expr)).head,sourceRowHasNoMatch = resolveOnJoinedPlan(Seq(col(TARGET_ROW_PRESENT_COL).isNull.expr)).head,matchedConditions = matchedClauses.map(clauseCondition),matchedOutputs = matchedClauses.map(matchedClauseOutput),notMatchedConditions = notMatchedClauses.map(clauseCondition),notMatchedOutputs = notMatchedClauses.map(notMatchedClauseOutput),noopCopyOutput =resolveOnJoinedPlan(targetOutputCols :+ Literal.FalseLiteral :+ incrNoopCountExpr),deleteRowOutput =resolveOnJoinedPlan(targetOutputCols :+ Literal.TrueLiteral :+ Literal.TrueLiteral),joinedAttributes = joinedPlan.output,joinedRowEncoder = joinedRowEncoder,outputRowEncoder = outputRowEncoder)val outputDF =Dataset.ofRows(spark, joinedPlan).mapPartitions(processor.processPartition)(outputRowEncoder)logDebug("writeAllChanges: join output plan:\n" + outputDF.queryExecution)
这里会涉及到两个ROW的转换,两个ROW的 Schema 是不一致的,如果要涉及到两个ROW之间的转换的时候,而且spark.implicits._
也没对应的隐式参数的时候,就得自己构造ExpressionEncoder[Row]
,其实 说到底 spark序列化和反序列化用的都是Expression表达式
,下面就来分析一下这里的序列化和反序列化是怎么做的。
分析
直接上代码:
object RowEncoder {def apply(schema: StructType, lenient: Boolean): ExpressionEncoder[Row] = {val cls = classOf[Row]val inputObject = BoundReference(0, ObjectType(cls), nullable = true)val serializer = serializerFor(inputObject, schema, lenient)val deserializer = deserializerFor(GetColumnByOrdinal(0, serializer.dataType), schema)new ExpressionEncoder[Row](serializer,deserializer,ClassTag(cls))}...
}
经过serializerFor
方法以后,返回 CreateNamedStruct(Seq(GetExternalRowField(BoundReference(0, ObjectType(cls), nullable = true),index,name)))
,注意如果,存在String类型的话,在序列化的时候会调用 StaticInvoke(classOf[UTF8String],"fromString")
进行反射调用序列化。
而经过deserializerFor
方法以后,返回CreateExternalRow(Seq(GetStructField(GetColumnByOrdinal(0, serializer.dataType))))
,注意对于 String类型的,在反序列化的时候会调用Invoke("toString")
反射调用反序列化。
而真正在进行行处理的时候,会调用ExpressionEncoder[Row].createSerializer
和ExpressionEncoder[Row].createDeserializer
。对于ExpressionEncoder[Row].createDeserializer
调用之前,还得调用resolveAndBind
进行参数的绑定。
对于序列化
主要是如下方法:
def createSerializer(): Serializer[T] = new Serializer[T](optimizedSerializer)
class Serializer[T](private val expressions: Seq[Expression])extends (T => InternalRow) with Serializable {@transientprivate[this] var inputRow: GenericInternalRow = _@transientprivate[this] var extractProjection: UnsafeProjection = _override def apply(t: T): InternalRow = try {if (extractProjection == null) {inputRow = new GenericInternalRow(1)extractProjection = GenerateUnsafeProjection.generate(expressions)}inputRow(0) = textractProjection(inputRow)} catch {case e: Exception =>throw QueryExecutionErrors.expressionEncodingError(e, expressions)}}
可以看到在apply的方法中会进行如下操作:
// 新建一个只有一列数据的ROW,并赋值为输入的值。
inputRow = new GenericInternalRow(1)
inputRow(0) = t
这里就和序列化的表达式BoundReference(0, ObjectType(cls), nullable = true)
吻合了: 取行数据中第一列的值.extractProjection 最终会根据 表达式计算出结果并返回 UnsafeRow
对于反序列化
在反序列化的时候,得先调用resolveAndBind
方法,进行Schema的绑定,便于从一样数据中取对应的数据。
def resolveAndBind(attrs: Seq[Attribute] = schema.toAttributes,analyzer: Analyzer = SimpleAnalyzer): ExpressionEncoder[T] = {val dummyPlan = CatalystSerde.deserialize(LocalRelation(attrs))(this)val analyzedPlan = analyzer.execute(dummyPlan)analyzer.checkAnalysis(analyzedPlan)val resolved = SimplifyCasts(analyzedPlan).asInstanceOf[DeserializeToObject].deserializerval bound = BindReferences.bindReference(resolved, attrs)copy(objDeserializer = bound)}
这个CatalystSerde.deserialize
方法获取deserializer
变量:
val deserializer: Expression = {if (isSerializedAsStructForTopLevel) {// We serialized this kind of objects to root-level row. The input of general deserializer// is a `GetColumnByOrdinal(0)` expression to extract first column of a row. We need to// transform attributes accessors.objDeserializer.transform {case UnresolvedExtractValue(GetColumnByOrdinal(0, _),Literal(part: UTF8String, StringType)) =>UnresolvedAttribute.quoted(part.toString)case GetStructField(GetColumnByOrdinal(0, dt), ordinal, _) =>GetColumnByOrdinal(ordinal, dt)case If(IsNull(GetColumnByOrdinal(0, _)), _, n: NewInstance) => ncase If(IsNull(GetColumnByOrdinal(0, _)), _, i: InitializeJavaBean) => i}} else {// For other input objects like primitive, array, map, etc., we deserialize the first column// of a row to the object.objDeserializer}}
这里会把表达式变成CreateExternalRow(Seq(GetColumnByOrdinal(index)))
,最终会
得到DeserializeToObject(UnresolvedDeserializer(CreateExternalRow(Seq(GetColumnByOrdinal(index)))),LocalRelation(attrs))
计划。
该计划经过ResolveDeserializer 规则解析, 会把 GetColumnByOrdinal(index)
变成对应的属性值。
最终 BindReferences.bindReference(resolved, attrs)
转换成Seq(BoundReference(ordinal, a.dataType, input(ordinal).nullable))
可执行表达式,最最终绑定到表的特定属性上,从而获取对应的值。
真正时机进行操作的时候,调用的是createDeserializer
方法:
def createDeserializer(): Deserializer[T] = new Deserializer[T](optimizedDeserializer)
class Deserializer[T](private val expressions: Seq[Expression])extends (InternalRow => T) with Serializable {@transientprivate[this] var constructProjection: Projection = _override def apply(row: InternalRow): T = try {if (constructProjection == null) {constructProjection = SafeProjection.create(expressions)}constructProjection(row).get(0, anyObjectType).asInstanceOf[T]} catch {case e: Exception =>throw QueryExecutionErrors.expressionDecodingError(e, expressions)}}
可以看到 最终的表达式CreateExternalRow(Seq(BoundReference(ordinal, a.dataType, input(ordinal).nullable)))
会生成 GenericRowWithSchema
类型的ROW,
constructProjection = SafeProjection.create(expressions)
constructProjection(row).get(0, anyObjectType).asInstanceOf[T]
其中 constructProjection 返回的是 SpecificInternalRow
类型的ROW。
所以constructProjection(row)
返回的是SpecificInternalRow(GenericRowWithSchema)
的值,所以get(0)是 GenericRowWithSchema类型的ROW,也就是ROW类型。
额外的话
对于BoundReference(ordinal, a.dataType, input(ordinal).nullable)
该方法,该方法是用来把表示涉及的属性,给映射到对应的计划的属性值上,这样我们计算的时候,就可以获取到对应的值,一般是调用BindReferences.bindReference
方法,这也是为什么表达式能获取到对应的属性值的原因。
相关文章:

Spark关于 ExpressionEncoder[T] 以及 RowEncoder 以及Spark怎么做外部数据和内部数据的转换
背景 本文基于 Spark 3.3, 最近在用 Spark Api 做 DataSet[Row] 转换的时候遇到了一些 Spark内部转换的问题, 在此记录一下。 杂谈 我们知道在Spark中分为 InternalRow和Row, 前者是 Spark 内部的使用的一行数据的表示,后者是给Spark开发者使用的行数…...

D-Day 上海站回顾丨以科技赋能量化机构业务
5月31日下午,DolphinDB 携手光大证券,在上海成功举办 D-Day 行业交流会。三十余位来自私募机构的核心策略研发、量化交易员、数据分析专家们齐聚现场,深入交流量化投研交易过程中的经验、挑战及解决方案。 DolphinDB 赋能机构业务平台 来自光…...

业财一体化的重点、难点和模式
业财一体化的内涵是企业将经营活动、财务管理、经营决策等进行科学的融合和管理,进而提高企业经营管理和财务决策的科学性,同时,基于IT技术、流程再造和组织重构更好的保障企业价值创造功能的实现。其涵盖管理循环、业务循环、信息循环三个双…...

gorse修改开源项目后,如何使用Docker compose发布
代码修改 git checkout v0.4.15 修改代码后提交。 镜像构建 export GOOSlinux export GOARCHamd64 export GOMAXPROCS8go build -ldflags"-s -w -X github.com/zhenghaoz/gorse/cmd/version.Version$(git describe --tags $(git rev-parse HEAD)) -X github.com/zhengh…...

Bowyer-Watson算法
数学原理及算法过程 Delaunay 三角剖分是一种特殊的三角剖分方法,它满足以下两个重要性质: 最大化最小角性质:Delaunay 三角剖分通过避免细长的三角形来最大化所有三角形的最小角。空外接圆性质:在 Delaunay 三角剖分中…...

计算机基础之:fork进程与COW机制
在Unix-like操作系统中,fork()是一个系统调用,用于创建一个与调用进程(父进程)几乎完全相同的新进程(子进程),包括父进程的内存空间、环境变量、文件描述符等。这个过程是通过写时复制ÿ…...

47.各种类型的线程池
线程池继承体系 Executor(interface)->ExecutorService(interface)->ThreadPoolExecutor(class) Executors.newFixedThreadPool 核心线程数最大线程数(没有救急线程被创建),所以也无需超时时间阻塞队列LinkedBlockingQueue,可以放任意…...

多目标优化-NSGA-II
文章目录 一、前置知识NSGA-II帕累托前沿 二、算法流程1.NSGA2.NSGA-II 一、前置知识 1.NSGA(非支配排序遗传算法):旨在同时优化多个冲突的目标函数,寻找帕累托前沿上的解集。 什么是多个冲突的目标: 比如你看上了一辆车,你既想要它便宜,又…...

元宇宙数字藏品交易所,未来发展的大趋势
随着科技的飞速进步,元宇宙以其独特的魅力为数字世界绘制了一幅前所未有的宏伟蓝图。在这一宏大的背景下,数字藏品交易所作为连接虚拟与现实的桥梁,正以其卓越的优势,引领着数字藏品市场迈向新的高度。 首先,元宇宙为…...

通配符https数字证书260
随着越来越多的人开始使用互联网,互联网上的信息变得繁杂,用户很难识别网站信息的真实性,为了维护互联网的环境,开发者开始使用https证书对网站传输数据进行加密和身份认证,以此来保护用户的隐私以及标示网站的真实性。…...

C++ | Leetcode C++题解之第133题克隆图
题目: 题解: class Solution { public:Node* cloneGraph(Node* node) {if (node nullptr) {return node;}unordered_map<Node*, Node*> visited;// 将题目给定的节点添加到队列queue<Node*> Q;Q.push(node);// 克隆第一个节点并存储到哈希…...

yangwebrtc x86_64环境搭建
版本:5.0.099 sudo apt-get install libxext-dev sudo apt-get install x11proto-xext-dev sudo apt-get install libxi-dev sudo apt install libasound2-dev sudo apt install libgl1-mesa-dev sudo apt-get install libxtst-dev 用qt打开以下两个项目的.pro met…...

前端面试题日常练-day53 【面试题】
题目 希望这些选择题能够帮助您进行前端面试的准备,答案在文末 1. 在PHP中,以下哪个函数可以用于从一个数组的末尾删除一个元素并返回被删除的元素? a) array_pop() b) array_push() c) array_shift() d) array_unshift() 2. 在PHP中&…...

空间不够用了怎么办
空间告急啊哥们 整理一下清理空间有用的一些blog吧。 【linux】公共服务器如何清理过多的.cache缓存 linux根目录空间不足,追加空间到根目录下 【linux】linux磁盘空间 目录查看清理 和 文件查看清理...

pytorch数学操作
文章目录 1.torch.bitwise_not()2.torch.bitwise_and()3.torch.ceil()3.torch.clamp()4.torch.torch.floor() 1.torch.bitwise_not() 在 PyTorch 中,torch.bitwise_not() 是一个函数,用于执行逐元素的位非(bitwise NOT)操作。 t…...

如何做好电子内窥镜的网络安全管理?
电子内窥镜作为一种常用的医疗器械,其网络安全管理对于保护患者隐私和医疗数据的安全至关重要。以下是一些基本原则和步骤,用于确保电子内窥镜的网络安全: 1. 数据加密 为了防止数据泄露,电子内窥镜在传输患者图像数据时应采取有…...

Spring Boot项目中,如何在yml配置文件中读取maven pom.xml文件中的properties标签下的属性值
一、前言 在最近的项目开发过程中,有一个需求,需要在Spring Boot项目的yml配置文件中读取到mave的 pom.xml文件中的properties标签下的属性值,这个要怎么实现呢? 二、技术实践 pom.xml文件中增加测试属性 <properties><…...

C++:模板进阶
✨✨✨学习的道路很枯燥,希望我们能并肩走下来! 文章目录 文章目录 前言 一 非类型模板参数 二 模板的特化 2.1 概念 2.2 函数模板特化 函数模板的易错点 2.3 类模板特化 2.3.1 全特化 2.3.2 偏特化 部分特化 参数更进一步的限制 2.3.3 类模板特化应用示例…...

Linux 磁盘分区步骤
1.lsblk用于查看磁盘分区情况,lsblk -f用于查看uuid字符串以及挂载点。 以下是虚拟机部分添加磁盘的步骤。 其余没展示的都按照默认设置进入下一步即可。 2.添加完成后使用reboot重新进入后再使用lsblk就会发现磁盘sdb已经有了,但是没有分区。现在添加分…...

【TB作品】 51单片机8x8点阵显示滚动汉字仿真
功能 题目5基于51单片机LED8x8点阵显示 流水灯 直接滚动显示HELLO 直接滚动显示老师好 代码 void main( void ) {/** 移位后,右边的是第一个595,接收0X02,显示出0X02* 移位后,左边的是第2个595,接收0Xfe,…...

c++简略实现共享智能指针Shared_Ptr<T>
重点: 1.引用计数在堆上(原本应为原子变量) 2.引用计数增加减少需要加锁保证线程安全。 3.内部实现Release函数用于释放资源 4.未实现,增加自定义删除器可以将Release修改为模板函数,传入可调用参数。对于shared_p…...

2024会声会影全新旗舰版,下载体验!
在当今数字时代,视频内容已成为最受欢迎的媒介之一。无论是个人娱乐、教育还是商业推广,优秀的视频制作都是吸引观众的关键。为了满足广大用户对高质量视频制作软件的需求,我们隆重推出了会声会影2024最新旗舰版。这款软件不仅集成了最先进的…...

使用 Node.js 和 Azure Function App 自动更新 Elasticsearch 索引
作者:来自 Elastic Jessica Garson 维护最新数据至关重要,尤其是在处理频繁变化的动态数据集时。这篇博文将指导你使用 Node.js 加载数据,并通过定期更新确保数据保持最新。我们将利用 Azure Function Apps 的功能来自动执行这些更新…...

UE4_Ben_图形52_水下效果处理
学习笔记,不喜勿喷,欢迎指正,侵权立删!祝愿生活越来越好! 在这个后期处理的效果中,我们可以看到有很多不同的,这里有浓雾,波纹扭曲,镜头扭曲和边缘模糊,在第4…...

RabbitMQ小结
MQ分类 Acitvemq kafka 优点:性能好,吞吐量高百万级,分布式,消息有序 缺点:单机超过64分区,cpu会飙高,消费失败不支持重试 , Rocket 阿里的mq产品 优点:单机吞吐量也…...

中国自动气象站:现代气象观测的中流砥柱
引言 气象观测是人类认识和预报天气的重要手段。在现代科技的推动下,自动气象站成为气象观测的重要工具,为天气预报、防灾减灾和气候研究提供了宝贵的数据支持。本文将介绍中国自动气象站的发展历程、技术特点及其在气象观测中的重要作用。 中国自动气象…...

【微信小程序】连接蓝牙设备
1、检查小程序是否授权蓝牙功能 initBluetooth() {const that thiswx.getSetting({success: (res) > {if (res.authSetting.hasOwnProperty(scope.bluetooth)) {//scope.bluetooth属性存在,且为falseif (!res.authSetting[scope.bluetooth]) {wx.showModal({tit…...

基于R语言BIOMOD2 及机器学习方法的物种分布模拟与案例分析实践技术
BIOMOD2是一个R软件包,用于构建和评估物种分布模型(SDMs)。它集成了多种统计和机器学习方法,如GLM、GAM、SVM等,允许用户预测和分析物种在不同环境条件下的地理分布。通过这种方式,BIOMOD帮助研究者评估气候…...

Objective-C之通过协议提供匿名对象
概述 通过协议提供匿名对象的设计模式,遵循了面向对象设计的多项重要原则: 接口隔离原则:通过定义细粒度的协议来避免实现庞大的接口。依赖倒置原则:高层模块依赖于抽象协议,而不是具体实现。里氏替换原则࿱…...

C语言基础(一)
C语言基础 一、标准输出(格式化输出):1、概念:2、注意语法点:3、格式控制符:4、调试技巧:5、代码风格:6、实例: 二、数据类型:1、整型概念:语法&a…...