hivehook 表血缘与字段血缘的解析
创始人
2025-01-15 23:03:38
0

代码

import org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext; import org.apache.hadoop.hive.ql.hooks.HookContext; import org.apache.hadoop.hive.ql.hooks.LineageInfo; import org.apache.hadoop.hive.metastore.api.Table; import org.slf4j.Logger; import org.slf4j.LoggerFactory;  import java.util.*;  public class HiveServerQueryLogHook implements ExecuteWithHookContext { 	static final Logger LOG = LoggerFactory.getLogger(HiveServerQueryLogHook.class); 	     @Override     public void run(HookContext hookContext) throws Exception {         printLineageInfo(hookContext);     }     private void printLineageInfo(HookContext hookContext) {         // 输出表         Set inputTables = new HashSet<>();;          // 输入表         Set outputTables = new HashSet<>();;          // 字段血缘 Map         // key为输出字段,value为来源字段数组         Map> fieldLineage = new HashMap<>(); 		 		// 从 `hookContext` 中获取 `Linfo` 并返回其 entry set,这意味着我们会获取到一个包含键值对的集合;遍历 `hookContext` 中 `Linfo` 的 entry set         for(Map.Entry dep: hookContext.getLinfo().entrySet()){             // 表血缘             // 将 `dep.getKey()` 转换为一个 Optional 对象,以防止空指针异常。             Optional.ofNullable(dep.getKey())             		// 如果 `dep.getKey()` 不为空,则将其转换为 `DataContainer` 对象。                     .map(LineageInfo.DependencyKey::getDataContainer)                     // 如果 `DataContainer` 不为空,则获取其表信息。                     .map(LineageInfo.DataContainer::getTable)                     // 将表信息传递给 `dealOutputTable` 方法进行处理。                     .map(this::dealOutputTable)                     // 如果处理后的结果不为空,则将其添加到 `outputTables` 集合中。                     .ifPresent(outputTables::add);             Optional.ofNullable(dep.getValue())                     .map(LineageInfo.Dependency::getBaseCols)                     .ifPresent(items -> items.stream().map(LineageInfo.BaseColumnInfo::getTabAlias)                             .map(LineageInfo.TableAliasInfo::getTable)                             .map(this::dealOutputTable)                             .forEach(inputTables::add));              // 字段血缘             // 将 `dep.getKey()` 转换为一个 Optional 对象,以防止空指针异常。             String column = Optional.ofNullable(dep.getKey())             		// 如果 `dep.getKey()` 不为空,则将其传递给 `dealDepOutputField` 方法进行处理。                     .map(this::dealDepOutputField)                     // 如果处理后的结果不为空,则将其作为键放入 `fieldLineage` 中,并关联一个空的 ArrayList,然后将该键返回给 `column` 变量。                      .map(aimField -> {                         fieldLineage.put(aimField, new ArrayList<>());                         return aimField;                     // 如果处理后的结果为空,则将 `column` 设置为 null。                     }).orElse(null);                     // 将 `dep.getValue()` 转换为一个 Optional 对象,以防止空指针异常。             Optional.ofNullable(dep.getValue())             		// 如果 `dep.getValue()` 不为空,则获取其基础列信息。                      .map(LineageInfo.Dependency::getBaseCols)                     // 如果基础列信息不为空,则将其转换为流并依次处理,将处理后的结果添加到 `fieldLineage` 中对应 `column` 的列表中。                     .ifPresent(items -> items.stream()                             .map(this::dealBaseOutputField)                             .forEach(item -> {                                 fieldLineage.get(column).add(item);                             }));         }         LOG.info("inputTables : {} ",inputTables);         LOG.info("outputTables : {} ",outputTables);         LOG.info("fieldLineage : {} ",fieldLineage.toString());     }     // 处理表的格式为 库.表     private String dealOutputTable(Table table) {         String dbName = table.getDbName();         String tableName = table.getTableName();         return dbName != null ? String.format("%s.%s", dbName, tableName) : tableName;     }      // 处理输出字段的格式     private String dealDepOutputField(LineageInfo.DependencyKey dependencyKey) {         try{             String tableName = dealOutputTable(dependencyKey.getDataContainer().getTable());             String field = dependencyKey.getFieldSchema().getName();             return String.format("%s.%s", tableName, field);         }catch (Exception e) {             LOG.error("deal dep output field error" + e.getMessage());             return null;         }     }      // 处理来源字段的格式     private String dealBaseOutputField(LineageInfo.BaseColumnInfo baseColumnInfo) {         try{             String tableName = dealOutputTable(baseColumnInfo.getTabAlias().getTable());             String field = baseColumnInfo.getColumn().getName();             return String.format("%s.%s", tableName, field);         }catch (Exception e) {             LOG.error("deal base output field error" + e.getMessage());             return null;         }     }  

配置

编译后生成jar文件添加到hive运行环境,设置hook

1)jar放置/disk1/hive-jars/hook  2)设置env,conf/hive-env.sh pushd /disk1/hive-jars/hook export HOOK_DEPS=$(ls *.jar| xargs -Ixx echo "`pwd`/xx" | sort | tr '\n' ':') popd export HIVE_AUX_JARS_PATH=${xxx%:}:${HOOK_DEPS%:}  3)修改hive-site.xml        hive.exec.post.hooks     com.xxx.xxx.HiveServerQueryLogHook            Comma-separated list of post-execution hooks to be invoked for each statement.       A post-execution hook is specified as the name of a Java class which implements the       org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext interface.          

测试

sql

use db_dev; CREATE TABLE IF NOT EXISTS `all_report_creator`( `project_id` INT COMMENT '项目组id', `report_id` INT COMMENT '报告id', `creator_id` INT COMMENT '报告创建者id', `nick` STRING COMMENT 'nick名字' )  STORED AS PARQUET; insert overwrite table all_report_creator select t1.project_id,t1.id,t2.id,t2.nick from db.new_report t1 left join db.bigviz_user t2 on t1.creator_id = t2.id where t1.project_id in(7,24) 

血缘

24/07/12 13:34:31 INFO xxx.HiveServerQueryLogHook: inputTables : [db.new_report, db.bigviz_user] 24/07/12 13:34:31 INFO xxx.HiveServerQueryLogHook: outputTables : [db_dev.all_report_creator] 24/07/12 13:34:31 INFO xxx.HiveServerQueryLogHook: fieldLineage : {db_dev.all_report_creator.project_id=[db.new_report.project_id], db_dev.all_report_creator.report_id=[db.new_report.id], db_dev.all_report_creator.creator_id=[db.bigviz_user.id], db_dev.all_report_creator.nick=[db.bigviz_user.nick]} 

参考

HIVE源码学习-hivehook尝试表血缘与字段血缘的解析
http://ganjiacheng.cn/article/2020/article_16_HIVE%E6%BA%90%E7%A0%81%E5%AD%A6%E4%B9%A0-hivehook%E5%B0%9D%E8%AF%95%E8%A1%80%E7%BC%98%E8%A7%A3%E6%9E%90/

相关内容

热门资讯

新2023版软件!Wepoke... 新2023版软件!Wepoke最新版外挂软件透明挂助手(透视)原来确实真的是有挂(有挂实测)-哔哩哔...
2022新版插件!Wepoke... 2022新版插件!Wepoke渠道外挂软件透明挂脚本(透视)其实确实是真的有挂(有挂经验)-哔哩哔哩...
新2023透明挂!Wepoke... 自定义新版Wepoke辅助器系统规律,只需要输入自己想要的开挂功能,一键便可以生成出Wepoke辅助...
新2023版软件!Wepoke... 新2023版软件!Wepoke必备外挂辅助器插件(透视)其实确实是有挂(有挂教会)-哔哩哔哩;AI智...
新2025版软件!Wepoke... 新2025版软件!Wepoke稳赢外挂辅助器神器(透视)原来确实是真的有挂(有挂技巧)-哔哩哔哩;是...
新2023透明挂!Wepoke... 新2023透明挂!Wepoke德州外挂辅助器脚本(透视)原来是有挂(有挂盘点)-哔哩哔哩新2023透...
阿里云注册域名怎么样(如何在阿... 阿里云注册域名怎么样?随着互联网技术的飞速发展,注册一个域名已经成为了网站建设的必经之路。而对于初次...
新2022版攻略!Wepoke... 您好,Wepoke输赢这款游戏可以开挂的,确实是有挂的,需要了解加微【136704302】很多玩家在...
阿里云怎么做网站(阿里云提供的... 阿里云怎么做网站在现代信息时代,拥有一个自己的网站越来越成为了一种必要。但是对于很多没有编程经验的人...
阿里云怎么提现的(阿里云提现操... 阿里云怎么提现的(阿里云提现操作流程详解)阿里云提供了非常便捷的账户提现功能,允许用户在满足一些简单...