我这边写了MyHive类,继承LineageLogger类,实现ExecuteWithHookContext接口
package com.zsins.hook;
import com.zsins.hook.utils.HiveSqlBloodFigureUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.ql.QueryPlan;
import org.apache.hadoop.hive.ql.hooks.*;
import org.apache.hadoop.hive.ql.optimizer.lineage.LineageCtx;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.*;
import java.util.Set;
public class MyHiveHook extends LineageLogger implements ExecuteWithHookContext {
private static Logger logger = LoggerFactory.getLogger(MyHiveHook.class);
// private static final HashSet<String> OPERATION_NAMES = new HashSet<>();
// static {
// // 建表
// OPERATION_NAMES.add(HiveOperation.CREATETABLE.getOperationName());
// // 修改数据库属性
// OPERATION_NAMES.add(HiveOperation.ALTERDATABASE.getOperationName());
// // 修改数据库属主
// OPERATION_NAMES.add(HiveOperation.ALTERDATABASE_OWNER.getOperationName());
// // 修改表属性,添加列
// OPERATION_NAMES.add(HiveOperation.ALTERTABLE_ADDCOLS.getOperationName());
// // 修改表属性,表存储路径
// OPERATION_NAMES.add(HiveOperation.ALTERTABLE_LOCATION.getOperationName());
// // 修改表属性
// OPERATION_NAMES.add(HiveOperation.ALTERTABLE_PROPERTIES.getOperationName());
// // 表重命名
// OPERATION_NAMES.add(HiveOperation.ALTERTABLE_RENAME.getOperationName());
// // 列重命名
// OPERATION_NAMES.add(HiveOperation.ALTERTABLE_RENAMECOL.getOperationName());
// // 更新列,先删除当前的列,然后加入新的列
// OPERATION_NAMES.add(HiveOperation.ALTERTABLE_REPLACECOLS.getOperationName());
// // 创建数据库
// OPERATION_NAMES.add(HiveOperation.CREATEDATABASE.getOperationName());
// // 删除数据库
// OPERATION_NAMES.add(HiveOperation.DROPDATABASE.getOperationName());
// // 删除表
// OPERATION_NAMES.add(HiveOperation.DROPTABLE.getOperationName());
// }
private static String toJson(Entity entity) throws Exception {
ObjectMapper mapper = new ObjectMapper();
switch (entity.getType()) {
case DATABASE:
Database db = entity.getDatabase();
return mapper.writeValueAsString(db);
case TABLE:
return mapper.writeValueAsString(entity.getTable().getTTable());
}
return null;
}
public void run(HookContext hookContext) {
// logger.info("this is my hive hook");
try{
assert (hookContext.getHookType() == HookContext.HookType.POST_EXEC_HOOK);
QueryPlan plan = hookContext.getQueryPlan();
LineageCtx.Index index = hookContext.getIndex();
SessionState ss = SessionState.get();
String operationName = plan.getOperationName();
logger.info("执行的SQL语句: " + plan.getQueryString());
LineageUtils lep = new LineageUtils();
lep.getLineageInfo(plan.getQueryString());
logger.info("Input tables = " + lep.getInputTableList());
logger.info("Output tables = " + lep.getOutputTableList());
logger.info("with tables = " + lep.getWithTableList());
if(!lep.getInputTableList().isEmpty()&&!lep.getOutputTableList().isEmpty()&&lep.getInputTableList().size()!=0&&lep.getOutputTableList().size()!=0){
HiveSqlBloodFigureUtil.parseAndCrud(plan.getQueryString());
}else{
logger.info("输出表或输入表为空,无需解析");
}
if(ss != null && index != null && !plan.isExplain() ) {
plan.getOutputs();
// if (OPERATION_NAMES.contains(operationName) && !plan.isExplain()) {
// logger.info("监控SQL操作");
// Set<ReadEntity> inputs = hookContext.getInputs();
Set<WriteEntity> outputs = hookContext.getOutputs();
//
for (Entity entity : inputs) {
logger.info("Hook metadata输入值: " + toJson(entity));
}
//
for (Entity entity : outputs) {
logger.info("Hook metadata输出值: " + toJson(entity));
}
} else {
logger.info("不在监控范围,忽略该hook!");
}
} catch ( Exception e) {
}
}
}
在"执行的SQL语句"处可以拿到执行的整条sql语句,然后结合hive sql解析工具类,实现整条sql语句的解析
这个解析工具类为LineageUtils
package com.zsins.hook;
import org.apache.hadoop.hive.ql.lib.*;
import org.apache.hadoop.hive.ql.parse.*;
import java.io.IOException;
import java.util.*;
public class LineageUtils implements NodeProcessor {
// 存放输入表
TreeSet<String> inputTableList = new TreeSet<String>();
// 存放目标表
TreeSet<String> outputTableList = new TreeSet<String>();
//存放with子句中的别名, 最终的输入表是 inputTableList减去withTableList
TreeSet<String> withTableList = new TreeSet<String>();
public TreeSet getInputTableList() {
return inputTableList;
}
public TreeSet getOutputTableList() {
return outputTableList;
}
public TreeSet getWithTableList() {
return withTableList;
}
@Override
public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx, Object... nodeOutputs) throws SemanticException {
ASTNode pt = (ASTNode) nd;
switch (pt.getToken().getType()) {
//create语句
case HiveParser.TOK_CREATETABLE: {
String createName = BaseSemanticAnalyzer.getUnescapedName((ASTNode) pt.getChild(0));
outputTableList.add(createName);
break;
}
//insert语句
case HiveParser.TOK_TAB: {
// System.out.println(pt.getChildCount() + "tab");
String insertName = BaseSemanticAnalyzer.getUnescapedName((ASTNode) pt.getChild(0));
outputTableList.add(insertName);
// System.out.println("insertName " + insertName);
break;
}
//from语句
case HiveParser.TOK_TABREF: {
ASTNode tabTree = (ASTNode) pt.getChild(0);
String fromName = (tabTree.getChildCount() == 1) ? BaseSemanticAnalyzer.getUnescapedName((ASTNode) tabTree.getChild(0)) : BaseSemanticAnalyzer.getUnescapedName((ASTNode) tabTree.getChild(0)) + "." + tabTree.getChild(1);
inputTableList.add(fromName);
break;
}
// with.....语句
case HiveParser.TOK_CTE: {
for (int i = 0; i < pt.getChildCount(); i++) {
ASTNode temp = (ASTNode) pt.getChild(i);
String cteName = BaseSemanticAnalyzer.getUnescapedName((ASTNode) temp.getChild(1));
withTableList.add(cteName);
}
break;
}
}
return null;
}
public void getLineageInfo(String query) throws ParseException, SemanticException {
ParseDriver pd = new ParseDriver();
ASTNode tree = pd.parse(query);
while ((tree.getToken() == null) && (tree.getChildCount() > 0)) {
tree = (ASTNode) tree.getChild(0);
}
inputTableList.clear();
outputTableList.clear();
withTableList.clear();
Map<Rule, NodeProcessor> rules = new LinkedHashMap<Rule, NodeProcessor>();
Dispatcher disp = new DefaultRuleDispatcher(this, rules, null);
GraphWalker ogw = new DefaultGraphWalker(disp);
ArrayList topNodes = new ArrayList();
topNodes.add(tree);
ogw.startWalking(topNodes, null);
}
//进行测试,sql语句是瞎写的,但是语法是对的
public static void main(String[] args) throws IOException, ParseException, SemanticException {
//String query = "insert into qc.tables_lins_cnt partition(dt='2016-09-15') select a.x from (select x from cc group by x) a left join yy b on a.id = b.id left join (select * from zz where id=1) c on c.id=b.id";
// String query ="from (select id,name from xx where id=1) a insert overwrite table dsl.dwm_all_als_active_d partition (dt='main') select id group by id insert overwrite table dsl.dwm_all_als_active_d2 partition (dt='main') select name group by name";
String query = "CREATE TABLE dcl.dcl_cmp_crm_ddsupplier_feeback\n" +
"as\n" +
"select \n" +
" ddsupplier_feeback_id ,\n" +
" policy_no ,\n" +
" workorder_id ,\n" +
" user_name ,\n" +
" id_no ,\n" +
" policy_licenseno ,\n" +
" rescue_project ,\n" +
" rescue_name ,\n" +
" specification ,\n" +
" num ,\n" +
" use_num ,\n" +
" residue_num ,\n" +
" equity_id ,\n" +
" phone_no ,\n" +
" concrrnphone ,\n" +
" indent_status ,\n" +
" random_num ,\n" +
" sign_time ,\n" +
" supplier_no ,\n" +
" supplier_name ,\n" +
" sysource ,\n" +
" opercode ,\n" +
" opername ,\n" +
" createdate ,\n" +
" updatedate ,\n" +
" remake ,\n" +
" islocked ,\n" +
" use_time ,\n" +
" order_time ,\n" +
" apply_phone_no ,\n" +
" mile ,\n" +
" start_poi_name ,\n" +
" end_poi_name ,\n" +
" start_poi_address ,\n" +
" end_poi_address ,\n" +
" use_property ,\n" +
" seatcount ,\n" +
" payamount ,\n" +
" road_rescue_totalnum ,\n" +
" substitute_driver_totalnum ,\n" +
" behalf_inspection_totalnum ,\n" +
" vehicle_safety_totalnum ,\n" +
" reserve01 ,\n" +
" risk_code \n" +
" from ods.cmp_crm_ddsupplier_feeback";
LineageUtils lep = new LineageUtils();
lep.getLineageInfo(query);
System.out.println("Input tables = " + lep.getInputTableList());
System.out.println("Output tables = " + lep.getOutputTableList());
System.out.println("with tables = " + lep.getWithTableList());
}
}
实现结果如下:
文章浏览阅读408次。 在ubuntu 8.04下安装Oracle 11g2008年05月22日 星期四 11:02oracle 11g 数据库虽然提供了linux x86的版本,但是支持的linux版本只有Red Hat,Novell and Solaris 这几个,debian 和 ubuntu 不在支持之列,所以在ubuntu下安装就相对麻烦一些,请照着下文的方法一步一步的安装,不
文章浏览阅读166次。新东方在线中考网整理了《初一英语下册语法知识点全汇总》,供同学们参考。一. 情态动词can的用法can+动词原形,它不随主语的人称和数而变化。1. 含有can的肯定句:主语+can+谓语动词的原形+其他。2. 含有can的否定句:主语+can't+动词的原形+其他。3. 变一般疑问句时,把can提前:Can+主语+动词原形+其他? 肯定回答:Yes,主语+can。否定回答:No,主语+can't...._七年级下册计算机知识点
文章浏览阅读3k次。在平时开发中,可能会遇到UFUN函数没有的功能,比如创建PTP的加工程序(我目前没找到,哪位大神可以指点一下),可以使用Grip创建PTP,然后用UFUN函数UF_call_grip调用Grip程序。具体如下截图(左侧UFUN,右侧Grip程序):..._uf调用grip
文章浏览阅读156次。第一个:原生普通样式(随着主题不同,样式会变)第二个:原生普通样式-小icon第三个:自定义RatingBar 颜色第四个:自定义RatingBar DrawableRatingBar 各样式实现===============原生样式原生样式其实没什么好说的,使用系统提供的style 即可<RatingBarstyle="?android:attr/ratingBarStyleIndicator"android:layout_width=“wrap_cont.._ratingbar样式修改
文章浏览阅读4.6k次,点赞6次,收藏11次。安装vs2017:参考vs2017下载和安装。安装cmake3.12.3:cmake是一个工程文件生成工具。用户可以使用预定义好的cmake脚本,根据自己的选择(像是Visual Studio, Code::Blocks, Eclipse)生成不同IDE的工程文件。可以从它官方网站的下载页上获取。这里我选择的是Win32安装程序,如图所示:然后就是运行安装程序进行安装就行。配置glfw3...._vs2017的opengl环境搭建(完整篇)
文章浏览阅读976次。MLC NAND,UBIFS_ubifs warning
文章浏览阅读2.2k次。计算机系统的两种存储器形式介绍时间:2016-1-6计算机系统的存储器一般应包括两个部分;一个是包含在计算机主机中的主存储器,简称内存,它直接和运算器,控制器及输入输出设备联系,容量小,但存取速度快,一般只存放那些急需要处理的数据或正在运行的程序;另一个是包含在外设中的外存储器,简称外存,它间接和运算器,控制器联系,存取速度虽然慢,但存储容量大,是用来存放大量暂时还不用的数据和程序,一旦要用时,就..._计算机存储器系统采用的是主辅结构,主存速度快、容量相对较小,用于 1 分 程序,外
文章浏览阅读5.6k次。1. STEP 7(Simatic Manager):STEP 7或者Simatic Manager是西门子PLC编程最常用的软件开发环境。4. STEP 7 MicroWin:STEP 7 MicroWn是一款专门针对微型PLC(S7-200系列PLC)的编程软件,是Simatic Manager的简化版。如果需要与PLC系统配合使用,则需要与PLC编程工具进行配合使用。除了上述软件之外,西门子还提供了一些配套软件和工具,如PLC模拟器、硬件调试工具等,以帮助PLC编程人员快速地进行调试和测试。_西门子plc编程软件
文章浏览阅读36次。【代码】HashMap扩容。_hashma扩容
文章浏览阅读2.9k次。1mvn dependency:copy-dependencies2 项目右键 -> Maven -> Disable Maven Nature3 项目右键 -> Configure -> Convert to Maven Project_maven资源加载不全,怎么重新加载
文章浏览阅读527次。DMLDML的全称是Database management Language,数据库管理语言。主要包括以下操作:insert、delete、update、optimize。本篇对其逐一介绍INSERT数据库表插入数据的方式:1、insert的完整语法:(做项目的过程中将字段名全写上,这样比较容易看懂)单条记录插入语法:insert into table_name (column_name1,......_dml的全称是
文章浏览阅读136次。可以参考: http://git.oschina.net/jrain-group/ 组织下的Java Modbus支持库Modbus-系列文章1、虚拟成对串口(1)下载虚拟串口软件VSPD(可在百度中搜索)image.png(2)打开软件,添加虚拟串口。在设备管理中,看到如下表示添加成功。..._最好用的 modebus调试工具