spark-sql字段血缘实现

spark-sql字段血缘实现

背景

Apache Spark是一个开源的大数据处理框架，它提供了一种高效、易于使用的方式来处理大规模数据集。在Spark中，数据是通过DataFrame和Dataset的形式进行操作的，这些数据结构包含了一系列的字段（也称为列）。字段血缘是Spark中的一个关键概念，它帮助我们理解数据的来源和流向，从而更好地理解和控制数据处理过程。

字段血缘是指在数据处理过程中，一个字段的值是如何从源数据产生并传递给目标数据的。在Spark中，字段血缘是通过依赖关系进行管理的。每个字段都有一个或多个依赖关系，这些依赖关系定义了字段的值如何从其他字段或数据源产生。

前提

spark版本：2.4.3
使用语言：java+scala

技术实现

1. spark-sql的执行计划，了解如何实现字段血缘解析

在这里插入图片描述
一个sql会经历一些列的处理，最终生成spark-core的代码，提交到集群运行。
首先看一下一个简单的sql生成的逻辑执行计划长什么样子

insert into default.jy_test
select * from default.jy_test

未解析的逻辑执行计划：

'InsertIntoTable 'UnresolvedRelation `default`.`jy_test`, false, false
+- 'Project [*]+- 'UnresolvedRelation `default`.`jy_test`

解析后（analyzer）的逻辑执行计划：

InsertIntoHiveTable `default`.`jy_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, false, false, [id, name]
+- Project [id#0, name#1]+- SubqueryAlias `default`.`jy_test`+- HiveTableRelation `default`.`jy_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#0, name#1]

优化后（optimizer）的逻辑执行计划：

InsertIntoHiveTable `default`.`jy_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, false, false, [id, name]
+- HiveTableRelation `default`.`jy_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#0, name#1]

重点来了

  1. 所谓的逻辑执行计划就是一个树形结构
  2. 树形结构中的叶子结点，就是hive的表信息：库名、表名、字段信息，并且spark给每一个字段生成了一个唯一的id
  3. 树形结构中的非叶子只包含了字段信息，不包含库表信息

所以想要实现字段血缘，我们需要做的就是通过那个生成的唯一id去一层层的关联，当关联到叶子结点的时候，就找到了库名表名

2. 构建一颗与解析后的逻辑执行计划一模一样的树形结构

  1. 首先定义node对象，用来存放节点信息
public abstract class Node {private String name;private List<Column> columnList = new ArrayList<>();private List<Node> children = new ArrayList<>();private Node parentNode;private String graphId;
}
  1. 其次定义column对象，用来存放字段信息
public class Column {private String name;private Long exprId;private String ColumnType;private ArrayList<Column> child = new ArrayList<Column>();private String tableName;private String process;
}

  1. 根据spark-sql生成的逻辑执行计划，我们为每一个逻辑节点创建对应的结点，由于结点很多，我这里直接给个截图，源码会在文章最后提供出来
    在这里插入图片描述

  2. 解析spark-sql生成的解析后的逻辑执行计划
    首先获取逻辑执行计划，这里提供两种方式：
    1.通过spark-session获取，该方法可以用来做测试，非常的方便

     LogicalPlan logicalPlan = spark.sessionState().sqlParser().parsePlan(sql);LogicalPlan analyzer = spark.sessionState().analyzer().execute(logicalPlan);

    2.通过QueryExecution获得，这里贴个图，详情看源码
    在这里插入图片描述

  3. 解析spark生成的analyzer，构建我们自己的树形结构
    这里贴一下主要的逻辑，使用scala去递归解析抽象语法树会方便很多

def resolveLogicPlan(plan: LogicalPlan, root: Node): Unit = {plan match {case plan: InsertIntoHadoopFsRelationCommand =>val node = root.asInstanceOf[Root]node.setName(NodeType.INSERTINTOHIVETABLE.getName)val database: String = plan.catalogTable.get.identifier.database.getOrElse("default")val table: String = plan.catalogTable.get.identifier.tableval fullTableName = database + "." + tableplan.catalogTable.get.schema.foreach { field => {val column = new Column()column.setName(field.name)column.setTableName(fullTableName)node.getColumnList.add(column)}}resolveLogicPlan(plan.query, node)case plan: SaveIntoDataSourceCommand =>val table: String = plan.options.get("dbtable").getOrElse("")val url: String = plan.options.get("url").getOrElse("")val user: String = plan.options.get("user").getOrElse("")val password: String = plan.options.get("password").getOrElse("")// 定义匹配数据库名称的正则表达式模式val pattern: Regex = ".*://[^/]+/(\\w+)".r// 使用正则表达式进行匹配val dbNameOption: Option[String] = pattern.findFirstMatchIn(url).map(_.group(1))val fullTableName = dbNameOption.getOrElse("") + "." + tableval node = root.asInstanceOf[Root]node.setName(NodeType.SAVEINTODATASOURCECOMMAND.getName)// 连接mysql，根据库名表明获取字段列表val fieldsList = getFieldsListFromMysql(url, user, password, table)fieldsList.foreach { field => {val column = new Column()column.setName(field)column.setTableName(fullTableName)node.getColumnList.add(column)}}resolveLogicPlan(plan.query, node)case plan: InsertIntoHiveTable =>val node = root.asInstanceOf[Root]node.setName(NodeType.INSERTINTOHIVETABLE.getName)val database: String = plan.table.identifier.database.getOrElse("default")val table: String = plan.table.identifier.tableval fullTableName = database + "." + tablenode.setTableName(fullTableName)plan.table.schema.foreach { field => {val column = new Column()column.setName(field.name)column.setTableName(fullTableName)node.getColumnList.add(column)}}resolveLogicPlan(plan.query, node)case plan: Aggregate =>val node = new AggregateNode()insertNodeColumnsFromNamedExpression(node, plan.aggregateExpressions)node.setParentNode(root)root.getChildren.add(node)resolveLogicPlan(plan.child, node)case plan: Project =>val node = new ProjectNode()insertNodeColumnsFromNamedExpression(node, plan.projectList)node.setParentNode(root)root.getChildren.add(node)resolveLogicPlan(plan.child, node)case plan: LogicalRelation =>val node = new LogicalRelationNode()dfsLogicalRelation(plan, node)node.setParentNode(root)root.getChildren.add(node)case plan: HiveTableRelation =>val node = new LogicalRelationNode()dfsLogicalRelation(plan, node)node.setParentNode(root)root.getChildren.add(node)case plan: Filter =>val node = new FilterNode()node.setParentNode(root)node.setCondition(plan.condition.toString)root.getChildren.add(node)resolveLogicPlan(plan.child, node)case plan: Join =>val node = new JoinNode()node.setName(plan.joinType.toString + " " + node.getName)node.setParentNode(root)node.setCondition(plan.condition.toString)root.getChildren.add(node)resolveLogicPlan(plan.left, node)resolveLogicPlan(plan.right, node)case plan: Window =>val node = new WindowNode()insertNodeColumnsFromNamedExpression(node, plan.windowExpressions)node.setParentNode(root)root.getChildren.add(node)resolveLogicPlan(plan.child, node)case plan: Union =>val node = new UnionNode()node.setParentNode(root)root.getChildren.add(node)plan.children.foreach(resolveLogicPlan(_, node))case plan: SubqueryAlias =>val node = new SubqueryNode()node.setName(node.getName + " " + plan.name.toString())node.setParentNode(root)root.getChildren.add(node)resolveLogicPlan(plan.child, node)case plan: Generate =>val node = new GenerateNode()processGenerate(plan, node)node.setParentNode(root)root.getChildren.add(node)resolveLogicPlan(plan.child, node)case _ =>plan.children.foreach(resolveLogicPlan(_, root))}}

  1. 到这里，我们已经得到了自己的树形结构。接下来要通过唯一id进行关联，补充库表信息。
    不知道大家注意没有，我们在node对象中有一个方法，

    在这里插入图片描述
    这里用到了访问者设计模式，感兴趣的同学可以学习一下，在spark-sql源码中，同样用的是访问者设计模式。

这里主要说一下Visitor的定义及方法：processColumn方法主要是拿自己的ExprId和所有孩子结点的ExprId比较，如果相等的话，说明是同一个字段，那就表名复制过来。

public interface Visitor {void visit(Node node);default void processColumn(Node node) {for (Column column1 : node.getColumnList()) {for (Node nd : node.getChildren()) {for (Column column2 : nd.getColumnList()) {processColumn(column1, column2);}}}}default void processColumn(Column column1, Column column2) {List<Column> child = column1.getChild();child.forEach(ch -> processColumn(ch, column2));if (column1.getExprId().equals(column2.getExprId())) {if(column2.getTableName() != null) {column1.setTableName(column2.getTableName());} else {column1.getChild().addAll(column2.getChild());}}}
}

LineageVisitor是Visitor的实现类， 主要用来做模式匹配，不同的结点处理方式会有不同，感兴趣的同学看一下这块的代码。

public class LineageVisitor implements Visitor{@Overridepublic void visit(Node node) {switch (node.getClass().getSimpleName()) {case "FilterNode" :case "SubqueryNode" :case "JoinNode" : copyChildColumnToThis(node); break;case "WindowNode" :case "GenerateNode" : copyChildColumnToThisWithProcess(node); break;case "Root" : processColumn((Root)node); break;case "UnionNode" : processColumn((UnionNode)node); break;default : processColumn(node);}}@Overridepublic void processColumn(Node node) {node.getChildren().forEach( child -> child.accept(this));Visitor.super.processColumn(node);}public void copyChildColumnToThis(Node node) {node.getChildren().forEach( child -> child.accept(this));for (Node child : node.getChildren()) {node.getColumnList().addAll(child.getColumnList());}}public void copyChildColumnToThisWithProcess(Node node) {node.getChildren().forEach( child -> child.accept(this));Visitor.super.processColumn(node);for (Node child : node.getChildren()) {node.getColumnList().addAll(child.getColumnList());}}public void processColumn(Root node) {node.getChildren().forEach( child -> child.accept(this));if(node.getColumnList().size() > 0) {for (int i = 0; i < node.getChildren().get(0).getColumnList().size(); i++) {for (Node child : node.getChildren()) {node.getColumnList().get(i).getChild().add(child.getColumnList().get(i));}}}}public void processColumn(UnionNode node) {node.getChildren().forEach( child -> child.accept(this));int size = node.getChildren().get(0).getColumnList().size();for (Column column : node.getChildren().get(0).getColumnList()) {Column column1 = new Column();column1.setName(column.getName());column1.setExprId(column.getExprId());node.getColumnList().add(column1);}for (int i = 0; i < size; i++) {for (Node child : node.getChildren()) {node.getColumnList().get(i).getChild().add(child.getColumnList().get(i));}}}
}

成果展示

还是拿最开始的sql ,看一下最终生成的字段血缘

insert into default.jy_test select * from default.jy_test

在这里插入图片描述

最后

字段血缘实现起来还是比较困难的，需要了解spak-sql的底层原理和一些技巧。
这里方便大家使用、学习、交流，所以贡献自己的源码，仓库地址：https://gitee.com/chenxiaoliang0901/crock/tree/main

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

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

相关文章

在ASP.NET MVC中使用JQuery提供的弹出窗口（模态窗口）

在ASP.NET MVC中使用JQuery提供的弹出窗口（模态窗口）

在ASP.NET MVC中使用JQuery提供的弹出窗口&#xff08;模态窗口&#xff09; 原理 使用<div>图层灵活显示在浏览器的任何位置。默认情况下指定<div>不可见 引用 样式表 在JQuery的官方网站可以下载对应的css样式表。打开官网的样例页。 找到样式表引用路径 …
阅读更多...
flex弹性盒子常用的布局属性详解

flex弹性盒子常用的布局属性详解

想必大家在开发中经常会用到flex布局。而且还会经常用到 justify-content 属性实现分栏等等 接下来给大家分别讲一下 justify-content 的属性值。 以下是我敲的效果图大家可以清晰看出区别 space-between 属性值可以就是说两端对齐 space-evenly 属性值是每个盒子之间的…
阅读更多...
Requests库的接口测试实现

Requests库的接口测试实现

Requests库是在接口测试中被广泛运用的库&#xff0c;包括模拟请求的下发&#xff0c;请求相关配置和响应结果的获取&#xff0c;核心主体都是通过request库完成。在接口测试中使用非常频繁。 一、Requests库环境搭建 接口测试的核心从模拟请求开始。在Python中&#xff0c;通…
阅读更多...
15个等轴视图设计的电动车汽车无人机等PR剪辑素材视频制作元素

15个等轴视图设计的电动车汽车无人机等PR剪辑素材视频制作元素

包含15个等轴视图、等距视角电动车、汽车、无人机、沙漏、飞机等PR剪辑素材视频制作元素mogrt动画模板。 特征&#xff1a; 等距设计&#xff1b; 可以更改颜色&#xff1b; 分辨率&#xff1a;全高清&#xff08;19201080&#xff09;&#xff1b; 持续时间&#xff1a;15秒&a…
阅读更多...
IDEA+Git——项目分支管理

IDEA+Git——项目分支管理

IDEAGit——项目分支管理 1. 前言2. 基础知识点2.1. 分支区分2.2. Git 代码提交规范2.3. 四个工作区域2.4. 文件的四种状态2.5. 常用命令2.6 注重点 3. IDEA分支管理 1. 前言 在Git中&#xff0c;分支是项目的不同版本&#xff0c;当开始开发一个新项目时&#xff0c;主分支通常…
阅读更多...
白话编程-概述

白话编程-概述

前言: 编程作为一种纯理性的学科,时刻都要明白自己到底在干什么.程序方面的书很多,却难以有那种把复杂问题讲清楚,让人很容易理解的.笔者想建立一种程序方面简单的思维,便于理解和记忆. 目录 一.概述 二.面向过程和面向对象 一.概述 编程到底是在干什么? 1)编程大多数情况…
阅读更多...
使用命令行方式搭建uni-app + Vue3 + Typescript + Pinia + Vite + Tailwind CSS + uv-ui开发脚手架

使用命令行方式搭建uni-app + Vue3 + Typescript + Pinia + Vite + Tailwind CSS + uv-ui开发脚手架

使用命令行方式搭建uni-app Vue3 Typescript Pinia Vite Tailwind CSS uv-ui开发脚手架 项目代码以上传至码云&#xff0c;项目地址&#xff1a;https://gitee.com/breezefaith/uniapp-vue3-ts-scaffold 文章目录 使用命令行方式搭建uni-app Vue3 Typescript Pinia V…
阅读更多...
深入理解C/C++指针：从基本操作到复杂表达式

深入理解C/C++指针：从基本操作到复杂表达式

目录 代码展示&#xff1a; 示例1&#xff1a;指向数组结束位置之后的地址 示例2&#xff1a;结构体大小对指针运算的影响 示例3&#xff1a;访问数组元素的不同方式 示例4&#xff1a;逗号表达式在数组初始化中的应用 示例5&#xff1a;计算多维数组元素间的地址差值 示…
阅读更多...
【Python】不一样的Ansible（一）

【Python】不一样的Ansible（一）

不一样的Ansible——进阶学习 前言正文概念Ansible CorePlugins和Modules 插件插件类型编写自定义插件基本要求插件选项文档标准编写插件 添加一个本地插件注册为内置插件指定插件目录 其他一些技巧更改Strategy 结语 前言 Ansible 是一个极其简单的 IT 自动化引擎&#xff0c…
阅读更多...
【Unity】动态申请权限

【Unity】动态申请权限

1、AndroidManifest.xml在<application></application>内添加一行&#xff1a; <meta-data android:name"unityplayer.SkipPermissionsDialog" android:value"true" /> 作用&#xff1a;屏蔽应用启动时弹出申请权限弹窗&#xff08;危…
阅读更多...
签到积分--题解--c++--模拟

签到积分--题解--c++--模拟

题目描述 每天签到一个应用程序&#xff0c;至少可以得 11 分&#xff0c;若前一天也签到了&#xff0c;则当天得分为前一天的得分加 11&#xff0c;一天最多只能得 55 分。 给定一个由 Y 及 N 构成的字符序列&#xff0c;该序列表示小爱每天签到的情况&#xff0c;签到记为 …
阅读更多...
Windows下Redis5+可视化软件下载、安装和配置教程-2024年1月8日

Windows下Redis5+可视化软件下载、安装和配置教程-2024年1月8日

Windows下Redis5下载、安装和配置教程-2024年1月8日 一、下载二、安装三、配置环境四、配置可视化客户端 一、下载 redis是现在是没有对win系统版进行维护的&#xff0c;这个是大神完成的&#xff0c;目前是到5版本&#xff0c;选择Redis-x64-5.0.14.1.zip点击下载 下载地址&…
阅读更多...
pgAdmin和asdf postgres的安装

pgAdmin和asdf postgres的安装

安装pgAdmin&#xff1a; curl https://www.pgadmin.org/static/packages_pgadmin_org.pub | sudo apt-key addsudo sh -c echo "deb https://ftp.postgresql.org/pub/pgadmin/pgadmin4/apt/$(lsb_release -cs) pgadmin4 main" > /etc/apt/sources.list.d/pgadmi…
阅读更多...
Linux CentOS官方文档之U盘安装

Linux CentOS官方文档之U盘安装

准本U盘安装媒介 https://docs.centos.org/en-US/centos/install-guide/Making_Media_USB/ 在Mac OS X上制作安装USB媒介 1. 查看U盘 $ diskutil list2. umountdisk $ diskutil unmountDisk /dev/disknumber Unmount of all volumes on disknumber was successful3. 使用d…
阅读更多...
.NET 6中如何使用Redis

.NET 6中如何使用Redis

1、安装redis Redis在windows平台上不受官方支持&#xff0c;所以想要在window安装Redis就必须去下载windows提供的安装包。安装地址&#xff1a;https://github.com/tporadowski/redis/releases 2、在NueGet安装包 3、在appsettings.json文件里面添加Redis相关配置信息 &quo…
阅读更多...
MySQL之子查询、连接查询(内外)以及分页查询

MySQL之子查询、连接查询(内外)以及分页查询

一、案例&#xff08;接上一篇文章&#xff09; 09&#xff09;查询学过「张三」老师授课的同学的信息 -- 一共有两种方式 -- 第一种方式&#xff1a; SELECT s.*,c.cname,t.tname,sc.score FROMt_mysql_teacher t,t_mysql_course c,t_mysql_student s,t_mysql_score sc WHERE…
阅读更多...
鸿蒙设备-开发板基础学习（BearPi-HM Micro）

鸿蒙设备-开发板基础学习（BearPi-HM Micro）

theme: minimalism 每当学习一门新的编程语言或者上手一款新的开发板&#xff0c;在学习鸿蒙设备开发过程中&#xff0c;带大家写的第一个程序&#xff0c;通过这个程序&#xff0c;我们可以对鸿蒙设备开发的整个流程有一个初步的体验。BearPi-HM Micro开发板为例&#xff1a;…
阅读更多...
「MCU」SD NAND芯片之国产新选择优秀

「MCU」SD NAND芯片之国产新选择优秀

文章目录 前言 传统SD卡和可贴片SD卡 传统SD卡 可贴片SD卡 实际使用 总结 前言 随着目前时代的快速发展&#xff0c;即使是使用MCU的项目上也经常有大数据存储的需求。可以看到经常有小伙伴这样提问&#xff1a; 大家好&#xff0c;请问有没有SD卡芯片&#xff0c;可以…
阅读更多...
MongoDB高级集群架构设计

MongoDB高级集群架构设计

两地三中心集群架构设计 容灾级别 RPO & RTO RPO&#xff08;Recovery Point Objective&#xff09;&#xff1a;即数据恢复点目标&#xff0c;主要指的是业务系统所能容忍的数据丢失量。RTO&#xff08;Recovery Time Objective&#xff09;&#xff1a;即恢复时间目标&…
阅读更多...
ARM CCA机密计算架构软件栈（上）

ARM CCA机密计算架构软件栈（上）

一、简介 该博客描述了Arm机密计算架构（Arm CCA）的固件和软件组件。 在这个博客中，将学到如何： 列出组成Arm CCA软件栈的组件集了解Arm CCA引入新软件组件的原因了解Monitor和领域管理监视器（RMM）的角色了解如何创建和管理领域1.1 开始之前 假设熟悉AArch64异常模型、…
阅读更多...
最新文章

Copyright @ 2022~2023