通过本文档,可以快速的解决Hive在Altas字段级血缘没有生成的问题,并了解Hive数据血缘实现原理。更多元数据管理,数据血缘相关文章,可以关注后续的文章更新。 文档共分为5个部分,层级结构如下图所示。
一、Hive与Atlas集成全流程
Apache Atlas 为组织提供开放式元数据管理和治理功能,用以构建其数据资产目录,对 这些资产进行分类和管理,形成数据字典。并为数据分析师和数据治理团队,提供围绕这些 数据资产的协作功能。
Atlas的安装部署可以参考我之前的文章:数据治理之元数据管理的利器——Atlas入门宝典
这次我们直接来看Atlas与Hive的集成过程。
首先准备,Hive连接Atlas的Hook包。
可以采用源码打包的方式。
在HDP平台上,通常可以从/usr/hdp/3.1.5.0-152/atlas/hook/hive/atlas-hive-plugin-impl获取Atlas Hive Hook的所有jar包(包括依赖包)。
- 将 atlas-application.properties 配置文件,压缩加入到 atlas-plugin-classloader-2.0.0.jar 中
#必须在此路径打包,才能打到第一级目录下 cd /usr/local/src/atlas/apache-atlas-2.1.0/conf zip -u /usr/local/src/atlas/apache-atlas-2.1.0/hook/hive/atlas-plugin-classloader-2.1.0.jar atlas-application.properties
- 修改 hive-site.xml
<property> <name>hive.exec.post.hooks</name> <value>org.apache.atlas.hive.hook.HiveHook</value> </property>
- 修改 hive-env.sh 的 Gateway 客户端环境高级配置代码段(安全阀)
HIVE_AUX_JARS_PATH=/usr/local/src/atlas/apache-atlas-2.1.0/hook/hive
- 修改 HIVE_AUX_JARS_PATH
- 修改 hive-site.xml 的 HiveServer2 高级配置代码段(安全阀)
<property> <name>hive.exec.post.hooks</name> <value>org.apache.atlas.hive.hook.HiveHook</value> </property> <property> <name>hive.reloadable.aux.jars.path</name> <value>/usr/local/src/atlas/apache-atlas-2.1.0/hook/hive</value> </property>
- 修改 HiveServer2 环境高级配置代码段
HIVE_AUX_JARS_PATH=/usr/local/src/atlas/apache-atlas-2.1.0/hook/hive
需要将配置好的Atlas包发往各个hive节点后重启集群。
但是,很多同学在按该步骤操作完以后,字段级数据血缘并未生成。这是为什么呢?
二、 CDH6、Hive2.1无字段数据血缘问题修复
原来是Hive是生成元数据日志的一个bug,此bug描述的问题是,用如下语句操作Hive时:
create table t1(id int, name string); create table t2 as select * from t1;
字段血缘关系无法生成,也就是说源码中这段代码不能生效。
lInfo = hookContext.getLinfo() > for(Map.Entry<LineageInfo.DependencyKey, LineageInfo.Dependency> e : > lInfo.entrySet()) { > System.out.println("Col Lineage Key : " + e.getKey()); > System.out.println("Col Lineage Value: " + e.getValue()); > }
随后Hive也对该问题进行了修复,不过修复的版本是后续版本,所以前面的版本受到了一些影响。
该补丁为:HIVE-14706,如需要获取补丁,可以关注大数据流动,回复“HIVE-14706”获取。
影响的版本主要是 2.1.0和2.1.1,这个问题在2.2.0中进行了修复。
补丁修复后,列级别数据血缘就能正常显示了。
此外还有一些Atlas与Hive存在兼容性问题,本文基于Atlas2.1.0兼容CDH6.3.2部署。Hive版本为2.1.1.其他版本的问题不在此文档讨论。
为兼容Hive2.1.1,需要修改源码重新编译。
- 所需修改的项目位置:
apache-atlas-sources-2.1.0\addons\hive-bridge
①.org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java 577行
String catalogName = hiveDB.getCatalogName() != null ? hiveDB.getCatalogName().toLowerCase() : null;
改为:
String catalogName = null;
②.org/apache/atlas/hive/hook/AtlasHiveHookContext.java 81行
this.metastoreHandler = (listenerEvent != null) ? metastoreEvent.getIHMSHandler() : null;
改为:C:\Users\Desktop\apache-atlas-2.1.0-sources\apache-atlas-sources-2.1.0\addons
this.metastoreHandler = null;
三、Hive血缘的核心—钩子函数
Hive的钩子非常重要,首先来了解下Hive的执行过程。
如果Hive通过MapReduce作为计算引擎为例,具体处理流程如下:
- HQL解析生成AST语法树
Antlr定义SQL的语法规则,完成SQL词法和语法解析,将SQL转化为抽象语法树AST Tree
- 语法分析得到QueryBlock
遍历AST Tree,抽象出查询的基本组成单元QueryBlock
- 生成逻辑执行计划
遍历QueryBlock,翻译为执行操作树Operator Tree
- Logical Optimizer Operator进行逻辑优化
逻辑层优化器进行OperatorTree变换,合并不必要的ReduceSinkOperator,减少shuffle数据量
- 生成物理执行计划Task Plan
遍历Operator Tree,翻译为MapReduce任务
- 物理优化Task Tree,构建执行计划QueryPlan
物理层优化器进行MapReduce任务的变换,生成最终的执行计划
- 表以及其他操作鉴权
- 执行引擎执行
在Hive Query整个生命周期中,会有如下钩子函数被执行:
HiveDriverRunHook的preDriverRun
该钩子函数由参数hive.exec.driver.run.hooks控制,决定要运行的pre hooks,多个钩子实现类以逗号间隔,钩子需实现 org.apache.hadoop.hive.ql.HiveDriverRunHook接口。
HiveSemanticAnalyzerHook的preAnalyze
在Driver开始run之前,HQL经过解析会进入编译阶段的语法分析,而在语法分析前会经过钩子HiveSemanticAnalyzerHook的preAnalyze方法处理。该钩子函数由hive.semantic.analyzer.hook配置,钩子需实现org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHook接口。
HiveSemanticAnalyzerHook的postAnalyze
与preAnalyze同属于一个钩子类,配置参数相同,会执行所有配置的语义分析hooks,但它位于Hive的语法分析之后,可以获取HQL的输入和输出表及分区信息,以及语法分析得到的task信息,由此可以判断是否是需要分布式执行的任务,以及执行引擎是什么。
生成执行计划之前的redactor钩子
该钩子由hive.exec.query.redactor.hooks配置,多个实现类以逗号间隔,钩子需继承org.apache.hadoop.hive.ql.hooks.Redactor抽象类,并替换redactQuery方法。
这个钩子函数是在语法分析之后,生成QueryPlan之前,所以执行它的时候语法分析已完成,具体要跑的任务已定,这个钩子的目的在于完成QueryString的替换,比如QueryString中包含敏感的表或字段信息,在这里都可以完成替换,从而在Yarn的RM界面或其他方式查询该任务的时候,会显示经过替换后的HQL。
task执行前的preExecutionHook
在执行计划QueryPlan生成完,并通过鉴权后,就会执行具体的task,而task执行之前会经过一个钩子函数,钩子函数由hive.exec.pre.hooks配置,多个钩子实现类以逗号间隔。实现方式:
1)实现org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext
通过实现该接口的run方法,执行所有的pre-execution hooks
// Pre/Post Execute Hook can run with the HookContext public interface ExecuteWithHookContext extends Hook { /** hookContext: The hook context passed to each hooks. * HookContext带有执行计划、Hive的配置信息、Lineage、UGI、提交的用户以及输入输出表等信息 */ void run(HookContext hookContext) throws Exception; }
2)实现org.apache.hadoop.hive.ql.hooks.PreExecute
该接口的run方法已经标注为过时,并且相对于ExecuteWithHookContext,PreExecute提供的信息可能不能完全满足我们的业务需求。
public interface PreExecute extends Hook { /** * The run command that is called just before the execution of the query. * SessionState、UGI、HQL输入表及分区信息,HQL输出表、分区以及本地和hdfs文件目录信息 */ @Deprecated public void run(SessionState sess, Set<ReadEntity> inputs,Set<WriteEntity> outputs, UserGroupInformation ugi) throws Exception; }
task执行失败时的ON_FAILURE_HOOKS
task执行失败时,Hive会调用这个hook执行一些处理措施。该钩子由参数hive.exec.failure.hooks配置,多个钩子实现类以逗号间隔。需实实现org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext接口。
task执行完成时的postExecutionHook
在task任务执行完成后执行。如果task失败,会先执行ON_FAILURE_HOOKS,之后执行postExecutionHook,该钩子由参数hive.exec.post.hooks指定的hooks(多个钩子实现类以逗号间隔)执行post execution hooks。实现方式:
1)实现org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext
2)实现org.apache.hadoop.hive.ql.hooks.PostExecute
ExecuteWithHookContext和PostExecute跟分别与上述task执行前的preExecutionHook、PreExecute对应,这里不再赘述。
HiveDriverRunHook的postDriverRun
在查询完成运行之后以及将结果返回给客户端之前执行,与preDriverRun对应。
此外,Hive中已经有一些内置实现的hook,下面举一些例子以及它们的主要作用:
ATSHook:实现了ExecuteWithHookContext,将查询和计划信息推送到Yarn App Timeline Server。
DriverTestHook:实现了HiveDriverRunHook的preDriverRun方法(对postDriverRun是空实现),用于打印输出的命令。
EnforceReadOnlyTables:pre execute hook,实现了ExecuteWithHookContext,用于阻止修改只读表。
LineageLogger:实现了ExecuteWithHookContext,它将查询的血统信息记录到日志文件中。LineageInfo包含有关query血统的所有信息。
PreExecutePrinter和PostExecutePrinter:pre和post hook的示例,它将参数打印输出。
PostExecTezSummaryPrinter:post execution hook,实现了ExecuteWithHookContext,可以打印Hive Tez计数器的相关信息。
PostExecOrcFileDump:post execution hook,实现了ExecuteWithHookContext,用于打印ORC文件信息。
UpdateInputAccessTimeHook:pre execution hook,可在运行查询之前更新所有输入表的访问时间。
特别强调一下LineageLogger和LineageInfo,对于做Hive血缘关系分析很有参考价值,这个下文会说。
通过对上面Hive中hook的执行"位置"和作用,以及Hive本身实现的一些Hook,分析可知:自定义hook,比如实现一个pre execution hook。
首先在maven的pom中引入hive-exec的依赖,如:
<dependency> <groupId>org.apache.hive</groupId> <artifactId>hive-exec</artifactId> <version>2.1.0</version> </dependency>
此外,还需创建一个实现ExecuteWithHookContext的类,实现其中的run方法,并设置相应的参数,使自定义的hook类生效。
四、Hive表数据血缘实现
表的实现就比较简单了。
Hive提供了org.apache.hadoop.hive.ql.tools.LineageInfo类,可以用来分析HiveQL中的表级别血缘关系。
public static void query(String[] args) throws IOException, ParseException, SemanticException { String query = args[0]; LineageInfo lep = new LineageInfo(); lep.getLineageInfo(query); for (String tab : lep.getInputTableList()) { System.out.println("InputTable=" + tab); } for (String tab : lep.getOutputTableList()) { System.out.println("OutputTable=" + tab); } }
如果我们调用该方法
import` `org.apache.hadoop.hive.ql.tools.LineageInfo; public` `class` `LineageInfoTest { public` `static` `void` `main(String[] args) ``throws` `Exception { String query = ``"INSERT OVERWRITE TABLE ccc PARTITION (dt='20221109') SELECT z.id AS id,z.name AS name FROM aaa z LEFT JOIN bbb c ON z.id = c.id AND z.dt='20221109' AND c.dt='20221109' ``; LineageInfo.query(``new` `String[] { query }); } }
将输出如下的结果:
InputTable=aaa InputTable=bbb OutputTable=ccc
五、Hive字段数据血缘实现原理
LineageLogger Hook 是Hive2.0版本
之后存在的,如果HIVE版本不够需要升级HIVE版本。
Hive提供了org.apache.hadoop.hive.ql.hooks.LineageLogger类,可以用来分析HiveQL中的字段级别血缘关系
具体设置如下
<property> ``<name>hive.``exec``.post.hooks<``/name``> ``<value>org.apache.hadoop.hive.ql.hooks.LineageLogger<``/value``> <``/property``>
hive.exec.post.hooks参数介绍
执行后置条件。一个用逗号分隔开的实现了org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext接口的java class列表,配置了该参数后,每个hiveQL语句执行后都要执行这个钩子,默认是空;
hive支持以下四种语句的血缘分析
- HiveOperation.QUERY
- HiveOperation.CREATETABLE_AS_SELECT
- HiveOperation.ALTERVIEW_AS
- HiveOperation.CREATEVIEW
配置hook输出
vim ${HIVE_HOME}/conf/hive-log4j2.properties og4j.logger.org.apache.hadoop.hive.ql.hooks.LineageLogger=INFO
输出位置在 hive-log4j2.properties 的 property.hive.log.dir 参数
测试输出
hive> desc test_table; OK c1 string c2 bigint c3 int hive> select c1, max(c2) as max_c2 from test_table group by c1;
输出结果为:
- vertices:顶点。代表参与DAG的节点元素,vertexType有COLUMN和TABLE两个值
- edges:边。代表DAG的流向,由sources指向targets,edgeType有PROJECTION(投影)和PREDICATE(谓语)两个值
{ "edges": [{ "sources": [2], "targets": [0], "edgeType": "PROJECTION" }, { "sources": [3], "targets": [1], "expression": "max(ods.test_table.c2)", "edgeType": "PROJECTION" }], "vertices": [{ "id": 0, "vertexType": "COLUMN", "vertexId": "c1" }, { "id": 1, "vertexType": "COLUMN", "vertexId": "max_c2" }, { "id": 2, "vertexType": "COLUMN", "vertexId": "ods.test_table.c1" }, { "id": 3, "vertexType": "COLUMN", "vertexId": "ods.test_table.c2" }] }
接下来就是将数据血缘存起来,然后进行展示了。
未完待续~
更多数据治理实践落地相关技术与资料,欢迎关注大数据流动。