《数据资产管理核心技术与应用》读书笔记-第三章：数据血缘

《数据资产管理核心技术与应用》是清华大学出版社出版的一本图书，全书共分10章，第1章主要让读者认识数据资产，了解数据资产相关的基础概念，以及数据资产的发展情况。第2～8章主要介绍大数据时代数据资产管理所涉及的核心技术，内容包括元数据的采集与存储、数据血缘、数据质量、数据监控与告警、数据服务、数据权限与安全、数据资产管理架构等。第9～10章主要从实战的角度介绍数据资产管理技术的应用实践，包括如何对元数据进行管理以发挥出数据资产的更大潜力，以及如何对数据进行建模以挖掘出数据中更大的价值。

图书介绍：数据资产管理核心技术与应用

今天主要是给大家分享一下第三章的内容：

第三章的标题为数据血缘

内容思维导图如下：

 1、获取数据血缘的技术实现

1.1、如何从Hive中获取数据血缘

Hive 是典型的数据仓库的代表，也是大数据中离线数据分层模型设计的代表，并且支持HQL的数据处理和计算，所以Hive为了方便用户做数据跟踪，在底层设计时，其实就考虑到了数据血缘跟踪这个问题。Hive 自身的血缘在其源码中主要通过org.apache.hadoop.hive.ql.hooks.LineageLogger.java 来输出，org.apache.hadoop.hive.ql.hooks.LineageLogger.java代码中主要处理的过程如下图所示，血缘主要通过edges（DAG图的流向）和vertices（DAG图的节点）来进行输出。

在org.apache.hadoop.hive.ql.hooks.LineageLogger.java的源码中定义了其支持的4种SQL操作类型，分别为QUERY（查询）、CREATETABLE_AS_SELECT（将Select 查询创建为一张表）、ALTERVIEW_AS（修改视图）、CREATEVIEW（创建视图）。

org.apache.hadoop.hive.ql.hooks.LineageLogger.java在解析和生成edges（DAG图的流向）和vertices（DAG图的节点）信息时，会判断QueryPlan（查询计划）的类型是否是支持的4种SQL操作类型中的一种，如果不是的话，就不会解析和生成edges和vertices。

org.apache.hadoop.hive.ql.hooks.Hook.java是Hive提供的Hook（钩子）功能，用于在Hive 任务执行前或者执行后，注入自定义的的操作代码。

具体的代码实现，可以直接参考纸质书，这里不再赘述。

1.2、 从Spark 执行计划中获取数据血缘

Spark底层生成执行计划以及处理执行计划的过程如下图所示。

从图中可以看到，

1、 执行SQL语句或者Data Frame时，会先生成一个Unresolved Logical Plan，就是没有做过任何处理和分析的逻辑执行计划，仅仅会从SQL语法的角度做一些基础性的校验。

2、 之后通过获取Catalog的数据，对需要执行的SQL语句做表名、列名的进一步分析校验，从而生成一个可以直接运行的逻辑执行计划。

3、 但是Spark底层会有个优化器来生成一个最优的执行操作方式，从而生成一个优化后的最佳逻辑执行计划。

4、 将最终确定下来的逻辑执行计划转换为物理执行计划，转换为最终的代码进行执行。

Spark的执行计划其实就是数据处理的过程计划，会将SQL语句或者DataFrame 做解析，并且结合Catalog一起，生成最终数据转换和处理的代码。所以可以从Spark的执行计划中，获取到数据的转换逻辑，从而解析到数据的血缘。但是spark的执行计划都是在spark底层内部自动处理的，如何获取到每次Spark任务的执行计划的信息呢？其实在Spark底层有一套Listener的架构设计，可以通过Spark Listener 来获取到spark 底层很多执行的数据信息。

在spark的源码中，以Scala的形式提供了一个org.apache.spark.sql.util.QueryExecutionListener  trait （类似Java 语言的接口），来作为Spark SQL等任务执行的监听器。在org.apache.spark.sql.util.QueryExecutionListener  中提供了如下表所示的两个方法。

方法名	描述
def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit	执行成功时，调用的方法，其中包括了执行计划参数，这里的执行计划可以是逻辑计划或者物理计划
def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit	执行失败时，调用的方法，其中同样也包括了执行计划参数，这里的执行计划可以是逻辑计划或者物理计划

因此可以借用QueryExecutionListener  来主动让Spark在执行任务时，将执行计划信息推送到自己的系统或者数据库中，然后再做进一步的解析，如下图

具体的代码实现，由于代码较多，请参考纸质书。

1.3、 从Spark  SQL语句中获取数据血缘

在Spark任务处理中，通过SQL语句来实现离线的ETL 数据处理是其在大数据中最常见的应用方式，如下图所示针对这种应用场景，可以直接通过获取Spark 中运行的SQL语句，然后通过解析SQL语句，并且结合Catalog，来分析出SQL 语句中包含的输入表和输出表的数据血缘关系。

有了通过SQL语句来解析血缘的思路后，需要解决的问题就是怎么自动抓取到Spark中运行的SQL语句。在上面的叙述中，有提到过Spark 有Listener的机制，org.apache.spark.sql.util.QueryExecutionListener只是Listener中的一种，可以通过Listener的方式自动获取到Spark的相关执行信息。在Spark中提供了org.apache.spark.scheduler.SparkListener这个底层抽象类来供上层代码监听Spark在整个生命执行周期中的相关事件消息，如下图所示。

获取Spark 执行的SQL 语句的整体流程总结如下图

具体的代码实现，由于代码较多，请参考纸质书。

1.4、 从Flink中获取数据血缘

FlinkSQL 在底层执行时，大概包含了如下的5个步骤，其底层执行过程和SparkSQL非常的类似。

• 第一步：对即将执行的SQL语句做语法分析，此时会通过Apache Calcite 将SQL语句直接转换为AST (Abstract Syntax Tree的简写，翻译过来就是抽象语法树），也就是Calcite 中的SqlNode节点树。Calcite是一个开源的SQL解析工具，Flink在底层技术实现时，集成了Calcite作为其底层的SQL语句解析器， Calcite的Github 地址为https://github.com/apache/calcite/，相关的更多介绍可以参考官方文档网址：https://calcite.apache.org/docs/。
• 第二步：根据查询到的元数据对SQL语句中的语法进行校验，通过第一步中获取到的SqlNode节点树中的信息来获取SQL语句中包含的表、字段、函数等信息，然后通过和元数据进行比对来判断SQL语句中包含的表、字段等信息在元数据中是否存在。
• 第三步：通过结合元数据信息对SqlNode节点树的进一步解析得到关系表达式树，生成初步的逻辑执行计划。
• 第四步：对第三步生成的逻辑执行计划进行进一步优化得到最优的逻辑执行计划，这一步得到结果还是关系表达式树。
• 第五步：将逻辑执行计划转变为物理执行计划，提交给Flink 集群进行执行。

具体的代码实现，由于代码较多，请参考纸质书。

1.5、   从数据任务的编排系统中获取数据血缘

        数据任务的编排系统通常是对不同的数据节点类型的任务进行前后运行顺序以及依赖关系的编排，如下图所示。

具体的代码实现，由于代码较多，请参考纸质书。

2、数据血缘的存储模型与展示设计

从架构设计的角度来看，血缘数据存储需要注意如下几点:

• 可扩展性：支持对数据血缘的源端进行扩展，比如当前只需要支持 Hive、Spark执行计划、Spark SQL等，但是未来可能会新增其他数据血缘来源。如果出现新的数据血缘来源时，要做到对现有的设计不做改动便可支持。
• 可跟踪性：需要记录数据血缘的变更记录，方便将来做追踪，比如数据血缘出现变更时，需要将其变更的过程记录下来，而不是在数据血缘发生变化后，直接替换现有的已经采集入库的数据血缘关系。
• 可维护性：支持手动维护，比如支持人工维护数据血缘或者数据血缘采集错误时，支持人工修改。

数据血缘的采集和处理的过程通常如下图所示，实时获取原始数据，然后发送到类似Kafka这样的消息队列中，然后对原始数据进行解析，生成血缘数据，然后入库保存。

在血缘数据解析入库后，就可以对数据血缘做展示了，关于数据血缘的展示设计参考如下图所示，一般需要注意如下几点：

• 支持表级血缘，默认展示表级血缘关系，选择单个表时，还可以点击查看该表的上游血缘或者下游血缘。
• 详情中支持字段级血缘展示，点击图中表的字段详情时，可以继续展开显示字段级的血缘关系。
• 血缘关系中，支持点击查看血缘的解析来源，比如SparkSQL语句、Spark执行计划等。

从下图可以看到表与表之间以及字段与字段之间的血缘关系展示。当某一张表发生变更时，很容易的就知道对下游或者上游的哪些表和字段产生影响，从而可以加快很多问题的处理和定位。在使用某张表的数据时，也能追溯到该表的原始数据表以及经过了哪些中间表的处理，数据的链路变得非常清晰，对数据的使用者来说，产生了极大的帮助。