来源：之家技术

1. 背景

随着汽车之家主机厂业务的发展，数据资产日益庞大，数据仓库的构建也越来越复杂，在日常的数仓构建工作中，常遇到数据模型建设不规范无法治理，数据溯源困难，数据模型修改导致业务分析困难等难题，此类问题主要是由于数据血缘分析不足造成的，只有强化血缘关系，才能更好的服务于数仓建模及治理和最大程度的发挥数据价值。

血缘关系在数仓建模中扮演着核心的角色。通过深入解析SQL语句并对其进行分析，我们可以洞察出各种模型之间的依赖关系，从而形成一张精细的血缘关系图。这张图不仅展示了数据模型之间的连锁关系，也为数据流程的上游和下游提供了清晰的指示。

对于数仓模型的构建来说，血缘关系也具有重要的意义。通过了解模型之间的依赖关系，我们可以更好地评估模型的健壮性和稳定性。此外，血缘关系还可以帮助我们优化数据模型的构建过程，提高模型的质量和效率。

在数据质量控制和治理方面，血缘关系同样发挥着不可忽视的作用。通过血缘关系的追踪，我们可以清晰地了解到数据从源头到目的地的整个过程，从而更好地控制数据的质量。同时，血缘关系也可以帮助我们实现数据的可追溯性，为数据治理提供有效的溯源手段。这样，一旦出现问题，我们就可以迅速找出问题的源头，从而采取有效的措施进行治理和纠正。

总的来说，血缘关系在数仓建模中起着至关重要的作用，对于数据分析师、数仓模型的构建以及数据质量控制和治理都具有深远的影响。

2. 遇到的问题

数据资产血缘分析过程中遇到的问题如下：

1.数据资产命名不规范：因历史存在未明确的命名规则并未强制执行，使得研发人员命名随意，增加血缘解析的复杂性。

2.ETL任务调度依赖遗漏：ETL任务调度依赖遗漏是一个常见的问题，这可能导致血缘关系的不完整或不准确。有时候，ETL任务之间的依赖关系可能非常复杂，如果某个任务没有正确配置或者遗漏或者定时任务无法设置依赖，那么血缘关系图就会不完整。

3.ETL脚本类型众多：不同的ETL脚本类型可能带来解析的困难。例如：shell脚本、Python脚本、PySpark脚本和HQL等都有可能在ETL过程中使用，这无疑增加了血缘解析的复杂性。

4.HQL文件没有统一的编程规范：单个HQL文件中出现多个insert语句，查询的表名不带数据库名。随意的编程风格降低了可读性和可维护性。

5.应用层数据资产调用不清晰：应用层数据服务方式有多种且没有埋点数据，例如导出至数据库为接口提供数据，自助分析平台直接检索，Kylin查询等，无法将数据资产与应用情况建立联系。

这些问题的出现对数据资产的血缘分析带来困难和挑战，也影响了数据治理和优化的效果。

3. 分析思路

3.1

埋点分析

业务埋点是在应用程序中嵌入特定的代码，用于跟踪和记录用户行为和业务操作。通过业务埋点，可以获取到应用程序中各个业务表的使用情况和调用次数，从而解决应用层数据资产调用不清晰的问题。

a.埋点分析整体流程

流程图如下：

图1

如上图红色标识部分，对业务后端服务通过AOP（面向切面编程）方式，注入MyBatis 拦截插件，切面将MyBaits拦截插件从各模块抽出来，降低了对模块的耦合度。通过插件的方式将与业务无关的代码封装起来，减少了重复代码。将SQL拦截器关注点统一到一个插件管理，修改更新方便，不影响其他业务模块。并且通过插件方式，减少对原先模块的侵入性，增强其灵活度，可以随意通过增加或者删除SQL拦截器插件来控制是否此业务模块需要进行埋点。

b.使用方式

如果模块需要对业务表进行埋点统计，只需在项目中引入SQL拦截器插件即可。如下：

    com.autohome.index    common-sqlplugin    1.1-SNAPSHOT

c.SQL拦截器

首先指定了要拦截的方法是Executor类的update方法、query方法，在 intercept方法中，获取查询或者更新的SQL语句，解析SQL获取表信息，然后存入redis中，然后，我们通过调用 invocation.proceed()来执行原始方法。

具体实现部分代码如下：

...//拦截器注解@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();            }        }    }    }

3.2

ETL任务解析

通过ETL任务脚本解析，解决单独依靠任务依赖分析血缘造成的血缘不完整问题，从而将血缘补充完整。另外解决ETL脚本类型众多，不能将全部的依赖关系融入到血缘关系中。

a.任务解析流程

图2

任务解析分为非常规脚本任务和常规ETL脚本任务两种，常规ETL脚本任务为正常的hql脚本文件，解析流程为通过SQL语句解析出血缘关系，存入MYSQL数据库；非常规脚本任务有shell脚本，Python脚本，PySpark脚本等，常规任务解析失败的也都归为非常规任务，每天解析脚本任务时，将此类脚本任务汇总发送预警邮件，然后线下，根据不同类型任务解析脚本，程序识别不了的，人工手动处理，通过上述自动程序处理后，人工处理部分占比为5%以内。

b.任务解析方式

首先获取常规脚ETL任务脚本内容，经过SQL解析工具解析出表血缘关系，最后存入MYSQL，如果解析失败，则存入非常规任务结果表，最终统一收集处理。

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()

3.3

SQL语句分析

图3

在处理SQL语句时，我们可以将其主要分为两个部分：source table和target table。这两个部分的解析和处理方式可能会有所不同。

首先，source table部分，如果包含with块，那么我们可以通过关键字with分割SQL语句截取出with部分，对with块内的select语句进行正则解析，生成KV数据。在这个过程中，我们可以将with块的名称作为key，将解析后的tableName数组作为value。

with块解析代码如下：

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)

然后，我们还需要对select语句进行解析，得到tableName数组。在以上过程中，如果tableName数组中存在不包含db的tableName，我们需要解析use语句，完成db映射。最后，将解析的tableName数组中的with名称替换为相应的tableName。

对于target table部分，我们可以通过解析insert语句得到。具体来说，我们需要解析insert into和insert overwrite table语句，得到target tableName。

最后，我们将target table和source table进行关联。在这个过程中，我们需要将source table中的tableName数组和target tableName进行匹配，完成关联操作。

获取select语句中的表名：

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语句中包含use语句且存在不含数据库信息的表名，则实现数据库与表名映射。代码如下：

# 判断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            ...

解析得到的tableName数组中如果存在为with块名称，则替换with对应的tableName数组。代码如下：

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

4. 分析及治理效果

4.1

分析效果数据

为了有效地分析SQL解析后资产的血缘状况，我们通过制定一种量化指标——血缘覆盖率，来关注数据资产的血缘情况。这种度量标准可以帮助我们更好地了解数据资产之间的关系以及数据从源头到目标的流动情况。

血缘覆盖率是一种关键性能指标，用于衡量数据资产的血缘关系被解析和追踪到的程度。这个指标含义指当至少有一条血缘链路与资产相关时，该资产即被血缘覆盖，而被血缘覆盖的资产占所有关注资产的比例即为血缘覆盖率。

在实际操作中，我们可以通过以下步骤来计算血缘覆盖率：

1. 确定关注的资产：根据业务需求和数据治理策略，确定需要关注的数据资产范围。

2. 识别血缘关系：通过解析SQL语句，构建数据资产之间的血缘关系。

3.计算血缘覆盖率：将血缘覆盖的资产数量与所有关注资产的总量相除，得到血缘覆盖率的数值。

4.分析血缘覆盖率：将计算得出的血缘覆盖率与其他指标进行对比，例如数据资产的总量、血缘关系的数量等。通过这种比较，可以发现哪些数据资产的血缘关系未被覆盖，以及哪些血缘链路未涉及到目标资产。

通过制定和计算血缘覆盖率这种量化指标，我们可以更好地了解SQL解析后数据资产的血缘状况，进而优化数据治理策略、提升数据质量、加强数据安全。

下图为分析效果数据：

图4

4.2

治理数据

根据业务埋点，统计出表使用频次，通过与业务库中的全量业务表进行对比，分析出僵尸表（长时间没有维护，没有人使用的表），这部分表也占用大量的存储空间，造成资源浪费。所以定期自动化分析清理僵尸表，协同业务方共同确认，是否已经真正没有人使用，包括除业务产品外的其他使用场景，比如不定时出业务报告，业务人员自己配置的业务看板等，最终确定的僵尸表，清理时，首先将数据备份到备份库中，其次将原库中的表删除，等待1个月时间后，确定没有任何业务异常发生，则从备份库中物理删除。目前，每周大约清理出10张僵尸表，后续清理不断推进中。

5. 总结

数据资产血缘分析是一种用于理解和追踪数据资产之间关系的过程，其目的是帮助组织更好地管理和保护其数据资产。

以下是数据资产血缘分析的过程：

1.确定分析目标：首先需要明确血缘分析的目标，例如理解数据流、发现数据依赖关系、识别数据风险等。

2. 收集数据资产：收集需要进行血缘分析的数据资产，这可能包括各种数据表、ETL过程等。

3.资产关系解析：通过解析数据资产（如SQL查询、ETL过程等），理解它们之间的依赖关系。

4.分析血缘关系：基于血缘数据分析各数据资产之间的血缘关系，例如哪些数据表被其他表引用、哪些查询依赖于其他查询等。这种分析可以帮助发现潜在的数据依赖问题，例如循环依赖、单点故障、跨层引用等。

5.风险评估：通过血缘分析，可以评估数据资产的风险，例如某表被大量查询引用可能需要进行优化，某查询结果被其他大量查询引用可能存在性能风险等。

6. 生成血缘报告：根据血缘分析的结果，生成血缘报告。报告应清晰地描述各数据资产之间的血缘关系、发现的问题及建议的解决方案。

7.优化建议：基于血缘报告的结果，可以提出数据治理、优化或保护的建议，例如对数据库进行优化、改进ETL过程等。

总之，数据资产血缘分析是一个持续的过程，需要定期进行以发现和解决数据管理问题并在实践中不断总结经验和方法。此外，随着组织环境和数据资产的改变，血缘分析的结果和应用也需要不断更新和改进。

6. 后续规划

后续的重点工作将会放在数据质量保障、集成多数据源、构建血缘关系图谱等方向，帮助改进数据资产血缘分析结果和应用，提高组织对数据管理和使用的效率和效果。

以下是一些思路：

1.完善数据质量保障：血缘分析的准确性很大程度上取决于数据本身的质量。可以进一步制定严格的数据质量管理策略，包括数据规范、数据清洗、数据映射等，以确保数据的准确性和一致性。

2.集成多源数据：探索将不同来源的数据进行整合，包括内部数据、外部数据、第三方数据等，以便更全面地了解数据的来龙去脉。

3.持续优化数据血缘模型：随着业务复杂性和数据量的增长，数据血缘模型可能需要进行调整和优化。可以针对不同的数据类型和业务需求，探索更加灵活、可扩展的血缘模型，以适应不断变化的数据流和数据处理过程。

4.构建血缘关系图谱：将数据血缘关系可视化，可以更直观地展示数据之间的关系和影响。帮助用户更好地理解数据资产之间的关联和依赖关系。

建立血缘分析标准和规范：为了使血缘分析更具可操作性和一致性，可以制定相关的标准和规范，明确血缘分析的流程、方法、工具和技术要求。这样可以提高血缘分析的可信度和可靠性，并为后续的血缘分析改进提供参考和依据。

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