dataworks 生成表血缘依赖

1.背景：

工作中用到了阿里云的dataworks,业务需求还没有到购买更高级的版本必要，没法查看表的血缘依赖。需要自己去实现血缘依赖。

思路: maxcompute 提供Information_Schema元数据查询，可以其中TASKS_HISTORY 表，可以查询到任务执行的日志， sql任务对应的执行sql 放在了operation_text，存在执行的sql,可以通过自定义函数解析sql生成表的血缘关系，这里需要用到阿里的druid，来解析sql语法。

实现过程：

1.maxcompute 自定义函数sql_parse

这里我是使用java的方式实现，需要引入的依赖，，

<dependency><groupId>com.aliyun.odps</groupId><artifactId>odps-sdk-udf</artifactId><version>0.29.10-public</version>
</dependency>

<dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>1.2.8</version>
</dependency>

实现逻辑是，传入sql 代码片段，通过druid进行解析，会返回这里返回的是个字符串拼接的结果，为啥没返回json呢，json在sql炸裂的时候不是很方便，最终结果返回格式为

来源表->结果表，来源表->结果表

实现代码如下,


public class SqlParse extends UDF {/**** @param sql* @return*/public String evaluate(String sql) {List<SQLStatement> stmts = SQLUtils.parseStatements(sql, JdbcConstants.ODPS);SchemaStatVisitor statVisitor = SQLUtils.createSchemaStatVisitor(JdbcConstants.ODPS);List<String> tableRelation = new ArrayList<>();for (SQLStatement stmt : stmts) {stmt.accept(statVisitor);Map<TableStat.Name, TableStat> tables = statVisitor.getTables();if (tables != null) {String value = null;List<String> tableList = new ArrayList<>();for (Map.Entry<TableStat.Name, TableStat> nameTableStatEntry : tables.entrySet()) {TableStat stat = nameTableStatEntry.getValue();String name = nameTableStatEntry.getKey().getName();if (stat.getCreateCount() > 0 || stat.getInsertCount() > 0) {value = name;} else if (stat.getSelectCount() > 0) {tableList.add(name);}}if (StringUtils.isNotBlank(value)) {for (String table : tableList) {tableRelation.add(table + "->" + value);}}}}return tableRelation.isEmpty() ? null : String.join(",", tableRelation);}}

2，上传jar,进行函数注册，参考官方文档

MaxCompute UDF概述 - 云原生大数据计算服务 MaxCompute - 阿里云

3.编写sql，提取表依赖

这里的逻辑不一定符合每个人的业务逻辑，可以自己对应结果，是否满足，做出对应调整。

-- use project_name; -- 对应的项目名称
select split(table_relation_value,"->")[0] source_table,split(table_relation_value,"->")[1] sink_table,1 as value  from (select REPLACE(sql_parse(nvl(operation_text,'')),"project_name.",'')  as  table_relation from  Information_Schema.TASKS_HISTORY where ds='${bizdate}'and   task_type in ("SQL","SQLRT")  and sql_parse(nvl(operation_text,''))  is not null
)t1
lateral view explode(split(t1.table_relation,",")) b as   table_relation_value
where  table_relation_value not  REGEXP 'tmp_dwd_print_industry_keyword|tmp_dim_print_user_industry_relation|Information_Schema.TASKS_HISTORY'
group by source_table,sink_table

tip: 如果每个sql任务，只会对应一个结果表，可以直接用input_tables，和output_tables,不用进行sql解析。

实现过程这期间我查看了很多别人的方案，还是觉得druid比较好，它本身就支持maxcompute 的语法。以下是我找到的druid的相关文档

介绍 - 《Alibaba Druid v1.0 使用手册》 - 书栈网 · BookStack

推荐一个做到比较好的在线血缘依赖网站：https://sqlflow.gudusoft.com/#/