1. 树叶云首页
  2. 技术资讯

Spark SQL 字段血缘在 vivo 互联网的实践

运维••技术资讯

作者：vivo互联网服务器团队-Hao Guangshi

一、背景

字段血缘是在表处理的过程中将字段的处理过程保留下来。为什么会需要字段血缘呢?

有了字段间的血缘关系，便可以知道数据的来源去处，以及字段之间的转换关系，这样对数据的质量，治理有很大的帮助。

Spark SQL 相对于 Hive 来说通常情况下效率会比较高，对于运行时间、资源的使用上面等都会有较大的收益。

平台计划将 Hive 任务迁移到 Spark SQL 上，同时也需要实现字段血缘的功能。

二、前期调研

开发前我们做了很多相关调研，从中得知 Spark 是支持扩展的：允许用户对 Spark SQL 的 SQL 解析、逻辑计划的分析和检查、逻辑计划的优化、物理计划的形成等进行扩展。

该方案可行，且对 Spark 的源码没有改动，代价也比较小，确定使用该方案。

三、Spark SQL 扩展

3.1 Spark 可扩展的内容

SparkSessionExtensions是比较重要的一个类，其中定义了注入规则的方法，现在支持以下内容：

  • 【Analyzer Rules】逻辑计划分析规则
  • 【Check Analysis Rules】逻辑计划检查规则
  • 【Optimizer Rules.】 逻辑计划优化规则
  • 【Planning Strategies】形成物理计划的策略
  • 【Customized Parser】自定义的sql解析器
  • 【(External) Catalog listeners catalog】监听器

在以上六种可以用户自定义的地方，我们选择了【Check Analysis Rules】。因为该检查规则在方法调用的时候是不需要有返回值的，也就意味着不需要对当前遍历的逻辑计划树进行修改，这正是我们需要的。

而【Analyzer Rules】、【Optimizer Rules】则需要对当前的逻辑计划进行修改，使得我们难以迭代整个树，难以得到我们想要的结果。

3.2 实现自己的扩展

classExtralSparkExtensionextends (SparkSessionExtensions=>Unit) {
overridedefapply(spark: SparkSessionExtensions): Unit= {

//字段血缘
spark.injectCheckRule(FieldLineageCheckRuleV3)

//sql解析器
spark.injectParser { case (_, parser) =>newExtraSparkParser(parser) }

  }
}

上面按照这种方式实现扩展，并在 apply 方法中把自己需要的规则注入到 SparkSessionExtensions 即可，除了以上四种可以注入的以外还有其他的规则。要让 ExtralSparkExtension 起到作用的话我们需要在spark-default.conf下配置

spark.sql.extensions=org.apache.spark.sql.hive.ExtralSparkExtension

在启动 Spark 任务的时候即可生效。

注意到我们也实现了一个自定义的SQL解析器，其实该解析器并没有做太多的事情。只是在判断如果该语句包含insert的时候就将 SQLText(SQL语句)设置到一个为FIELD_LINE_AGE_SQL，之所以将SQLText放到FIELD_LINE_AGE_SQL里面。因为在 DheckRule 里面是拿不到SparkPlan的我们需要对SQL再次解析拿到 SprkPlan，而FieldLineageCheckRuleV3的实现也特别简单，重要的在另一个线程实现里面。

这里我们只关注了insert语句，因为插入语句里面有从某些个表里面输入然后写入到某个表。

classExtraSparkParser(delegate: ParserInterface) extendsParserInterfacewithLogging{

overridedefparsePlan(sqlText: String): LogicalPlan= {
vallineAgeEnabled=SparkSession.getActiveSession
      .get.conf.getOption("spark.sql.xxx-xxx-xxx.enable").getOrElse("false").toBoolean
logDebug(s"SqlText: $sqlText")
if(sqlText.toLowerCase().contains("insert")){
if(lineAgeEnabled){
if(FIELD_LINE_AGE_SQL_COULD_SET.get()){
//线程本地变量在这里
FIELD_LINE_AGE_SQL.set(sqlText)
        }
FIELD_LINE_AGE_SQL_COULD_SET.remove()
      }
    }
delegate.parsePlan(sqlText)
  }
//调用原始的sqlparser
overridedefparseExpression(sqlText: String): Expression= {
delegate.parseExpression(sqlText)
  }
//调用原始的sqlparser
overridedefparseTableIdentifier(sqlText: String): TableIdentifier= {
delegate.parseTableIdentifier(sqlText)
  }
//调用原始的sqlparser
overridedefparseFunctionIdentifier(sqlText: String): FunctionIdentifier= {
delegate.parseFunctionIdentifier(sqlText)
  }
//调用原始的sqlparser
overridedefparseTableSchema(sqlText: String): StructType= {
delegate.parseTableSchema(sqlText)
  }
//调用原始的sqlparser
overridedefparseDataType(sqlText: String): DataType= {
delegate.parseDataType(sqlText)
  }
}

3.3 扩展的规则类

caseclassFieldLineageCheckRuleV3(sparkSession:SparkSession) extends (LogicalPlan=>Unit ) {

valexecutor: ThreadPoolExecutor=
ThreadUtils.newDaemonCachedThreadPool("spark-field-line-age-collector",3,6)

overridedefapply(plan: LogicalPlan): Unit= {
valsql=FIELD_LINE_AGE_SQL.get
FIELD_LINE_AGE_SQL.remove()
if(sql!=null){
//这里我们拿到sql然后启动一个线程做剩余的解析任务
valtask=newFieldLineageRunnableV3(sparkSession,sql)
executor.execute(task)
    }

  }
}

很简单，我们只是拿到了 SQL 然后便启动了一个线程去得到 SparkPlan，实际逻辑在

FieldLineageRunnableV3。

3.4 具体的实现方法

3.4.1 得到 SparkPlan

我们在 run 方法中得到 SparkPlan：


overridedefrun(): Unit= {
valparser=sparkSession.sessionState.sqlParser
valanalyzer=sparkSession.sessionState.analyzer
valoptimizer=sparkSession.sessionState.optimizer
valplanner=sparkSession.sessionState.planner
............
valnewPlan=parser.parsePlan(sql)
PASS_TABLE_AUTH.set(true)
valanalyzedPlan=analyzer.executeAndCheck(newPlan)

valoptimizerPlan=optimizer.execute(analyzedPlan)
//得到sparkPlan
valsparkPlan=planner.plan(optimizerPlan).next()
...............
if(targetTable!=null){
vallevelProject=newArrayBuffer[ArrayBuffer[NameExpressionHolder]]()
valpredicates=newArrayBuffer[(String,ArrayBuffer[NameExpressionHolder])]()
//projection
projectionLineAge(levelProject, sparkPlan.child)
//predication
predicationLineAge(predicates, sparkPlan.child)
...............

为什么要使用 SparkPlan 呢?当初我们考虑的时候，物理计划拿取字段关系的时候是比较准的，且链路比较短也更直接。

在这里补充一下 Spark SQL 解析的过程如下：

经过SqlParser后会得到逻辑计划，此时表名、函数等都没有解析，还不能执行;经过Analyzer会分析一些绑定信息，例如表验证、字段信息、函数信息;经过Optimizer 后逻辑计划会根据既定规则被优化，这里的规则是RBO，当然 Spark 还支持CBO的优化;经过SparkPlanner后就成了可执行的物理计划。

我们看一个逻辑计划与物理计划对比的例子：

一个 SQL 语句：

selectitem_id,TYPE,v_value,imeifromt1
unionall
selectitem_id,TYPE,v_value,imeifromt2
unionall
selectitem_id,TYPE,v_value,imeifromt3

逻辑计划是这样的：

物理计划是这样的：

显然简化了很多。

得到 SparkPlan 后，我们就可以根据不同的SparkPlan节点做迭代处理。

我们将字段血缘分为两种类型：projection(select查询字段)、predication(wehre查询条件)。

这两种是一种点对点的关系,即从原始表的字段生成目标表的字段的对应关系。

想象一个查询是一棵树，那么迭代关系会如下从树的顶端开始迭代，直到树的叶子节点，叶子节点即为原始表：

那么我们迭代查询的结果应该为

id->tab1.id ,
name->tab1.name,tabb2.name,
age→tabb2.age。

注意到有该变量

vallevelProject=newArrayBuffer
[ArrayBuffer[NameExpressionHolder]]()，通过projecti-onLineAge迭代后levelProject
存储了顶层id，name,age对应的(tab1.id)，(tab1.name,tabb2.name)，(tabb2.age)。

当然也不是简单的递归迭代，还需要考虑特殊情况例如：Join、ExplandExec、Aggregate、Explode、GenerateExec等都需要特殊考虑。

例子及效果：

SQL:


withAas (selectid,name,agefromtab1whereid>100 ) ,
Cas (selectid,name,max(age) fromAgroupbyA.id,A.name) ,
Bas (selectid,name,agefromtabb2whereage>28)
insertintotab3
selectC.id,concat(C.name,B.name) asname, B.agefrom
B,CwhereC.id=B.id

效果：


{
"edges": [
    {
"sources": [
3
      ],
"targets": [
0
      ],
"expression": "id",
"edgeType": "PROJECTION"
    },
    {
"sources": [
4,
7
      ],
"targets": [
1
      ],
"expression": "name",
"edgeType": "PROJECTION"
    },
    {
"sources": [
5
      ],
"targets": [
2
      ],
"expression": "age",
"edgeType": "PROJECTION"
    },
    {
"sources": [
6,
3
      ],
"targets": [
0,
1,
2
      ],
"expression": "INNER",
"edgeType": "PREDICATE"
    },
    {
"sources": [
6,
5
      ],
"targets": [
0,
1,
2
      ],
"expression": "((((default.tabb2.`age` IS NOT NULL) AND (CAST(default.tabb2.`age` AS INT) > 28)) AND (B.`id` > 100)) AND (B.`id` IS NOT NULL))",
"edgeType": "PREDICATE"
    },
    {
"sources": [
3
      ],
"targets": [
0,
1,
2
      ],
"expression": "((default.tab1.`id` IS NOT NULL) AND (default.tab1.`id` > 100))",
"edgeType": "PREDICATE"
    }
  ],
"vertices": [
    {
"id": 0,
"vertexType": "COLUMN",
"vertexId": "default.tab3.id"
    },
    {
"id": 1,
"vertexType": "COLUMN",
"vertexId": "default.tab3.name"
    },
    {
"id": 2,
"vertexType": "COLUMN",
"vertexId": "default.tab3.age"
    },
    {
"id": 3,
"vertexType": "COLUMN",
"vertexId": "default.tab1.id"
    },
    {
"id": 4,
"vertexType": "COLUMN",
"vertexId": "default.tab1.name"
    },
    {
"id": 5,
"vertexType": "COLUMN",
"vertexId": "default.tabb2.age"
    },
    {
"id": 6,
"vertexType": "COLUMN",
"vertexId": "default.tabb2.id"
    },
    {
"id": 7,
"vertexType": "COLUMN",
"vertexId": "default.tabb2.name"
    }
  ]
}

四、总结

在 Spark SQL 的字段血缘实现中，我们通过其自扩展，首先拿到了 insert 语句，在我们自己的检查规则中拿到

SQL 语句，通过SparkSqlParser、Analyzer、Optimizer、SparkPlanner，最终得到了物理计划。

我们通过迭代物理计划，根据不同执行计划做对应的转换，然后就得到了字段之间的对应关系。当前的实现是比较简单的，字段之间是直线的对应关系，中间过程被忽略，如果想实现字段的转换的整个过程也是没有问题的。

文章来源网络，作者：运维，如若转载，请注明出处：https://shuyeidc.com/wp/299472.html<

赞 (0)
运维的头像运维
00
上一篇2025-05-23 10:20
下一篇 2025-05-23 10:21

相关推荐

  • 如何制作个人主题技术资讯

    个人主题怎么制作？

    制作个人主题是一个将个人风格、兴趣或专业领域转化为视觉化或结构化内容的过程，无论是用于个人博客、作品集、社交媒体账号还是品牌形象，核心都是围绕“个人特色”展开，以下从定位、内容规划、视觉设计、技术实现四个维度,详细拆解制作个人主题的完整流程，明确主题定位：找到个人特色的核心主题定位是所有工作的起点，需要先回答……

    运维的头像运维
    2025-11-20
    0
  • 如何管理社群营销技术资讯

    社群营销管理关键是什么？

    社群营销的核心在于通过建立有温度、有价值、有归属感的社群，实现用户留存、转化和品牌传播，其管理需贯穿“目标定位-内容运营-用户互动-数据驱动-风险控制”全流程,以下从五个维度展开详细说明：明确社群定位与目标社群管理的首要任务是精准定位，需明确社群的核心价值（如行业交流、产品使用指导、兴趣分享等）、目标用户画像……

    运维的头像运维
    2025-11-20
    0
  • 香港公司如何网站备案技术资讯

    香港公司网站备案需要什么材料？

    香港公司进行网站备案是一个涉及多部门协调、流程相对严谨的过程，尤其需兼顾中国内地与香港两地的监管要求，由于香港公司注册地与中国内地不同，其网站若主要服务内地用户或使用内地服务器，需根据服务器位置、网站内容性质等，选择对应的备案路径（如工信部ICP备案或公安备案），以下从备案主体资格、流程步骤、材料准备、注意事项……

    运维的头像运维
    2025-11-20
    0
  • 如何企业上云推广技术资讯

    如何企业上云推广

    企业上云已成为数字化转型的核心战略,但推广过程中需结合行业特性、企业痛点与市场需求，构建系统性、多维度的推广体系，以下从市场定位、策略设计、执行落地及效果优化四个维度，详细拆解企业上云推广的实践路径，精准定位：明确目标企业与核心价值企业上云并非“一刀切”的方案，需先锁定目标客户群体，提炼差异化价值主张，客户分层……

    运维的头像运维
    2025-11-20
    0
  • PS如何设计搜索框技术资讯

    PS设计搜索框的实用技巧有哪些？

    在PS中设计一个美观且功能性的搜索框需要结合创意构思、视觉设计和用户体验考量，以下从设计思路、制作步骤、细节优化及交互预览等方面详细说明，帮助打造符合需求的搜索框，设计前的规划明确使用场景：根据网站或APP的整体风格确定搜索框的调性，例如极简风适合细线条和纯色，科技感适合渐变和发光效果，电商类则可能需要突出搜索……

    运维的头像运维
    2025-11-20
    0

发表回复

您的邮箱地址不会被公开。必填项已用 * 标注