主机厂数据资产血缘分析治理实践

<dependency>    <groupId>com.autohome.index</groupId>    <artifactId>common-sqlplugin</artifactId>    <version>1.1-SNAPSHOT</version></dependency>

...//拦截器注解@Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class}),public class SqlPluginInterceptor implements Interceptor {         @Override    public Object intercept(Invocation invocation) throws Throwable {        MappedStatement mappedStatement = (MappedStatement) invocation.getArgs()[0];        Object parameter = null;        if (invocation.getArgs().length > 1) {            parameter = invocation.getArgs()[1];        }        BoundSql boundSql = mappedStatement.getBoundSql(parameter);        Configuration configuration = mappedStatement.getConfiguration();        long startTime = System.currentTimeMillis();        try {            return invocation.proceed();        } finally {            try {                 ......                //将查询sql和表信息存入redis                saveSqlToRedis(retSQL,sqlCost,mappedStatement.getId());            } catch (Exception e) {                System.out.println("sql拦截出错!");                e.printStackTrace();            }        }    }    }

try :        file_content = hql_parse_util.parse_hql(hqlfile)        table_dict= hql_parse_util.parse_output(file_content.lower())        for out_put_table in table_dict :            input_tables = table_dict[out_put_table]            if out_put_table == 'tbnames':                out_put_table = '.'            for input_table in input_tables:                if input_table.startswith('--'):                    continue                sql_val.append([hqlfile,out_put_table.split('.',1)[0],out_put_table.split('.',1)[1],                                input_table.split('.',1)[0],input_table.split('.',1)[1],uid,name,'hive',job_status,date_time])    cnt = cursor.executemany(insert_sql, sql_val)    print ('已入库常规任务' + str(cnt) + '条')    conn.commit()finally:    cursor.close()

def sql_parse(sql):    sql = sql.lower()    with_tb_dict = {}    sql = re.sub('with[-_a-z]','',sql)    if 'with' in sql:        with_all_arr = sql.split("with")[1:]        for with_content in with_all_arr:            # with 块解析            with_content = with_content[0:with_content.index("insert")].strip()            # 正则匹配with部分            with_content = re.sub('\\)\s*,\s*\'','',with_content)            with_arr =re.split('\\)\s*,', with_content,0)            with_arr = [content + ")" for content in with_arr[:-1]] + [with_arr[-1]]            with_list = []            # 遍历截取的with块语句 解析出with 别名与内部查询的表名关系            ...            # 解析多with块，查询引用的情况            ...        print (with_tb_dict)

def sql_parse_get_tablenames(sql_query):    table_names = re.findall(r'from\s+(\S+)', sql_query, re.IGNORECASE)    table_names += re.findall(r'join\s+(\S+)', sql_query, re.IGNORECASE)    tb_names = []    for name in table_names:        # 不包含 '('        if '(' not in name:            tb_names.append(name.replace(')','').replace('`','').replace(';',''))    result = list(set(tb_names))    return result

# 判断sql中是否包含useif 'use ' in sql:    use_arrs = sql.split('use ')[1:]    for arr in use_arrs:        db_name = arr.split(';')[0]        if 'insert' in arr:            # insert 解析            insert_arr = arr.split("insert ")[1:]            for tb in insert_arr:                if 'into' in tb:                    insert_tb = tb[tb.index("into ") + len("into "):tb.index("\n")].strip()                else:                    insert_tb = tb[tb.index("table ") + len("table "):tb.index("\n")].strip()                hive_target_tb = insert_tb.split(" ")[0]                if '.' not in hive_target_tb:                    hive_target_tb = db_name + '.' + hive_target_tb                tb_source_tb = []                # 映射tableName数组中with名                ...                # 映射target table 和 source table                ...        else:            tb_source_tb = []            # 映射tableName数组中with名            ...            # 映射target table 和 source table            ...

def sql_parse_tb_map(with_tb_dict,sql):    source_tb = sql_parse_get_tablenames(sql)    if len(with_tb_dict) > 0:        for s in range(len(source_tb)-1,-1,-1):            if source_tb[s] in with_tb_dict:                source_tb +=  with_tb_dict.get(source_tb[s])                source_tb.remove(source_tb[s])    return source_tb