正文共： 9053字 12图

预计阅读时间： 23分钟

前几天，Datahub提供了最新的字段级别数据血缘功能，很多朋友迫不及待想对比一下Datahub的字段级血缘与Atlas的区别。

这个时候问题来了，在Atlas收集Hive血缘的时候，由于部分版本问题，没有显示出字段级的数据血缘。这是为什么呢？其实只要做一个简单的修复就可以了，但是知其然也要知其所以然。今天我们就来看一下这个问题到底是怎么引起的，然后从HiveSql的语法树讲起，看看数据血缘到底是如何被检测到的。

最后提醒，文档版权为公众号 大数据流动 所有，请勿商用。相关技术问题以及安装包可以联系笔者独孤 风加入相关技术交流群讨论获取

另外，为了将我之前写作的文章，还有积累的资料留下来。去年的时候，我申请了知识星球《大数据流动资料库》。虽然想进行一些具体的分类，但是精力有限，以后将只维护这一个知识星球。包括我写作文章的PDF版本和收集的资料都会上传到这里，包括未来付出较多时间整理的付费文章，我也会同步到这里。

  • 星球分享各种相关资料，包括但不限于大数据，实时计算，数据治理，数据可视化，用户画像，及实现开源技术如Datahub，Superset，Atlas，Spark，Flink，Minio等等。

  • 星球为学习资料首发地，并将永久存储这些资料。

  • 星球为多个技术微信群问题解答汇总处；

  • 定期组织实战小组，共同学习进步；

  • 现在加入可以查看之前上传的所有内容，当然后期随着内容不断增加，需要购买一些知识星球的服务，为了更好的服务星友，价格会根据人数上调。

正文开始：

通过本文档，可以快速的解决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/confzip -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作为计算引擎为例，具体处理流程如下：

  1. HQL解析生成AST语法树

Antlr定义SQL的语法规则，完成SQL词法和语法解析，将SQL转化为抽象语法树AST Tree

  1. 语法分析得到QueryBlock

遍历AST Tree，抽象出查询的基本组成单元QueryBlock

  1. 生成逻辑执行计划

遍历QueryBlock，翻译为执行操作树Operator Tree

  1. Logical Optimizer Operator进行逻辑优化

逻辑层优化器进行OperatorTree变换，合并不必要的ReduceSinkOperator，减少shuffle数据量

  1. 生成物理执行计划Task Plan

遍历Operator Tree，翻译为MapReduce任务

  1. 物理优化Task Tree，构建执行计划QueryPlan

物理层优化器进行MapReduce任务的变换，生成最终的执行计划

  1. 表以及其他操作鉴权

  2. 执行引擎执行

在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                      inthive> 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"}]
}

接下来就是将数据血缘存起来，然后进行展示了。

未完待续~

更多数据治理实践落地相关技术与资料，欢迎关注大数据流动。

聊聊Hive数据血缘——从Atlas没有列级血缘的Bug讲起相关推荐

  1. Hive数据存储结构

    Hive数据存储: Hive没有专门的数据存储格式,也没有为数据建立索引,用户可以自由组织Hive中的表,只需要在创建表的时候告诉Hive数据中的分隔符(列分隔和行分隔),Hive就可以解析数据. h ...

  2. 使用Ranger对Hive数据进行脱敏

    Ranger支持对Hive数据的脱敏处理(Data Masking),它对select的返回结果进行脱敏处理,对用户屏蔽敏感信息. 备注: 该功能针对HiveServer2的场景(如beeline/j ...

  3. Hive数据导入——数据存储在Hadoop分布式文件系统中，往Hive表里面导入数据只是简单的将数据移动到表所在的目录中！...

    转自:http://blog.csdn.net/lifuxiangcaohui/article/details/40588929 Hive是基于Hadoop分布式文件系统的,它的数据存储在Hadoop ...

  4. Redis 小白指南（二）- 聊聊五大类型：字符串、散列、列表、集合和有序集合...

    Redis 小白指南(二)- 聊聊五大类型:字符串.散列.列表.集合和有序集合 引言 开篇<Redis 小白指南(一)- 简介.安装.GUI 和 C# 驱动介绍>已经介绍了 Redis 的 ...

  5. 一篇文章彻底明白Hive数据存储的各种模式

    Hive的数据分为表数据和元数据,表数据是Hive中表格(table)具有的数据;而元数据是用来存储表的名字,表的列和分区及其属性,表的属性(是否为外部表等),表的数据所在目录等.下面分别来介绍. 一 ...

  6. HIVE 数据倾斜浅谈

    HIVE 数据倾斜浅谈 一.数据倾斜现象   map100%,reduce一直卡在一个值,如99%. 二.数据倾斜的原因   数据按key的hash值分配到reduce中,如果有的key值比较集中,就 ...

  7. Hive数据倾斜优化

    Hive数据倾斜 Group By 中的计算均衡优化 1.Map端部分聚合 先看看下面这条SQL,由于用户的性别只有男和女两个值 (未知).如果没有map端的部分聚合优化,map直接把groupby_ ...

  8. hive数据倾斜及处理案例

    什么是数据倾斜 数据倾斜其实是进行分布式计算的时候,某些节点的计算能力比较强或者需要计算的数据比较少,早早执行完了,某些节点计算的能力较差或者由于此节点需要计算的数据比较多,导致出现其他节点的redu ...

  9. KUDU数据导入尝试一：TextFile数据导入Hive,Hive数据导入KUDU

    背景 SQLSERVER数据库中单表数据几十亿,分区方案也已经无法查询出结果.故:采用导出功能,导出数据到Text文本(文本>40G)中. 因上原因,所以本次的实验样本为:[数据量:61w条,文 ...

最新文章

  1. java mysql aio_Java中的NIO，BIO，AIO分别是什么
  2. 【Java 注解】自定义注解 ( 注解属性定义与赋值 )
  3. leetcode 378. Kth Smallest Element in a Sorted Matrix | 378. 有序矩阵中第 K 小的元素（小根堆）
  4. mysql回滚与错误提示_对mysql事务提交、回滚的错误理解 – jae – 博客园
  5. 从一个简单的Java单例示例谈谈并发
  6. data ajax begin,Ajax.BeginForm()知多少
  7. [转]电影《龙纹身女孩》中的那句 SQL-----The Girl With The ANSI Tattoo
  8. Android应用中使用及实现系统“分享”接口
  9. 新手产品经理怎样去选择一款原型工具
  10. 拓端tecdat|R语言对巨灾风险下的再保险合同定价研究案例：广义线性模型和帕累托分布Pareto distributions分析
  11. vs2010 快捷键大全
  12. C4D阿诺德Arnold渲染器介绍
  13. Bugku CTF 每日一题 想蹭网先解开密码
  14. 东芝打印机共享怎么设置_东芝打印机如何共享
  15. android 模拟器的使用
  16. 最新AI产品经理求职动态：卡年龄、卡学历，这么卷，怎么办？
  17. python 排产计划_生产排程计划表
  18. 文本识别OCR浅析：特征篇
  19. yyyy-MM-dd HH:mm:ss和时间戳之间的转换
  20. Linux树莓派开发——刷机

热门文章

  1. Jmeter压力测试，个人使用总结
  2. XP系统启动时滚动条总是时间很长
  3. 大数据介绍项目流程_大数据介绍
  4. 抓取、下载某位博主的豆瓣日记
  5. 【论文笔记】Catching Both Gray and Black Swans: Open-set Supervised Anomaly Detection*
  6. mongodb数据库自动备份
  7. pinbox: 我用过最好的收藏工具
  8. linux 串口转网口工具,linux下串口调试工具/串口终端推荐: picocom(转)
  9. qt 绘图工具（一）
  10. 玩转华为数据中心交换机系列 | 配合Eth-Trunk接口流量本地优先转发