sqllineage解析sql列级血缘并提交到datahub

这篇具有很好参考价值的文章主要介绍了sqllineage解析sql列级血缘并提交到datahub。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

版本信息

操作记录

安装datahub v0.10.0

datahub 快速部署

元数据摄取

通过sqlline获取指定sql文件中HiveSQL的字段级血缘关系,并将结果提交到datahub


版本信息

python 3.8.16

datahub v0.10.0

操作记录

安装datahub v0.10.0

sqllineage解析sql列级血缘并提交到datahub

详见datahub官网 A Metadata Platform for the Modern Data Stack | DataHub

执行命令

python3 -m pip install --upgrade pip wheel setuptools

python3 -m pip install --upgrade acryl-datahub==0.10.0

查看版本

python3 -m datahub version

sqllineage解析sql列级血缘并提交到datahub

 文章来源地址https://www.toymoban.com/news/detail-476284.html

datahub 快速部署

将datahub v0.10.0分支下的docker-compose-without-neo4j.quickstart.yml文件准备到本地

datahub/docker-compose-without-neo4j.quickstart.yml at v0.10.0 · datahub-project/datahub · GitHub


确保以下端口未被占用

  • 3306 for MySQL

  • 9200 for Elasticsearch

  • 9092 for the Kafka broker

  • 8081 for Schema Registry

  • 2181 for ZooKeeper

  • 9002 for the DataHub Web Application (datahub-frontend)

  • 8080 for the DataHub Metadata Service (datahub-gms)

 如有占用在命令行传参进行替换

datahub docker quickstart --mysql-port 53306

执行

python3 -m datahub docker quickstart -f ./docker-compose-without-neo4j.quickstart.yml --version v0.10.0

开始拉取镜像

sqllineage解析sql列级血缘并提交到datahub

 

成功构建容器,datahub启动成功

sqllineage解析sql列级血缘并提交到datahub

 

访问hadoop105:9002

sqllineage解析sql列级血缘并提交到datahub

 

输入账号、密码datahub

sqllineage解析sql列级血缘并提交到datahub

 

元数据摄取

安装hive插件

python3 -m pip install 'acryl-datahub[hive]'

安装过程中报错

sqllineage解析sql列级血缘并提交到datahub

 

尝试安装依赖项

yum -y install gcc gcc-c++ python-devel.x86_64 cyrus-sasl-devel.x86_64 gcc-c++.x86_64

再次安装hive插件

sqllineage解析sql列级血缘并提交到datahub

 

检查datahub插件

python3 -m datahub check plugins

hive插件成功安装

sqllineage解析sql列级血缘并提交到datahub

 

编写摄取hive元数据的配置文件

source:
  type: "hive"
  config: 
    host_port: "hadoop102:10000" # hiveserver2 

sink:
  type: "datahub-rest"
  config:
    server: "http://hadoop105:8080" # datahub gms server

开始摄取hive元数据

python3 -m datahub ingest -c ./hive-metadata-ingestion.yml

元数据摄取完成

sqllineage解析sql列级血缘并提交到datahub

 

进入web页面查看

sqllineage解析sql列级血缘并提交到datahub

sqllineage解析sql列级血缘并提交到datahub

 

sqllineage解析sql列级血缘并提交到datahub

 

 

通过sqllineage获取指定sql文件中HiveSQL的字段级血缘关系,并将结果提交到datahub

参考datahub官方文档给出的提交细粒度血缘的脚本datahub/lineage_emitter_dataset_finegrained.py at master · datahub-project/datahub · GitHub

参考sqllineage文档Getting Started — sqllineage 1.3.7 documentation

结合sqllineage,获取指定sql的列级血缘,再调用datahub rest api,将结果提交到datahub

具体py代码如下

from sqllineage.runner import LineageRunner

import datahub.emitter.mce_builder as builder
from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.com.linkedin.pegasus2avro.dataset import (
    DatasetLineageType,
    FineGrainedLineage,
    FineGrainedLineageDownstreamType,
    FineGrainedLineageUpstreamType,
    Upstream,
    UpstreamLineage,
)
import sys

'''
    解析目标sql文件的HiveSQL生成列级血缘,提交到datahub
    sql文件路径作为命令行参数传入脚本
    提交到datahub的platform = hive
'''


# 库名设置
def datasetUrn(tableName):
    return builder.make_dataset_urn("hive", tableName)  # platform = hive


# 表、列级信息设置
def fieldUrn(tableName, fieldName):
    return builder.make_schema_field_urn(datasetUrn(tableName), fieldName)


# 目标sql文件路径
sqlFilePath = sys.argv[1]

sqlFile = open(sqlFilePath, mode='r', encoding='utf-8')

sql = sqlFile.read().__str__()

# 获取sql血缘
result = LineageRunner(sql)

# 获取sql中的下游表名
targetTableName = result.target_tables[0].__str__()

print(result)

print('===============')

# 打印列级血缘结果
result.print_column_lineage()

print('===============')

# 获取列级血缘
lineage = result.get_column_lineage

# 字段级血缘list
fineGrainedLineageList = []

# 用于冲突检查的上游list
upStreamsList = []

# 遍历列级血缘
for columnTuples in lineage():
    # 上游list
    upStreamStrList = []

    # 下游list
    downStreamStrList = []

    # 逐个字段遍历
    for column in columnTuples:

        # 元组中最后一个元素为下游表名与字段名,其他元素为上游表名与字段名

        # 遍历到最后一个元素,为下游表名与字段名
        if columnTuples.index(column) == len(columnTuples) - 1:
            downStreamFieldName = column.raw_name.__str__()
            downStreamTableName = column.__str__().replace('.' + downStreamFieldName, '').__str__()

            # print('下游表名:' + downStreamTableName)
            # print('下游字段名:' + downStreamFieldName)

            downStreamStrList.append(fieldUrn(downStreamTableName, downStreamFieldName))
        else:
            upStreamFieldName = column.raw_name.__str__()
            upStreamTableName = column.__str__().replace('.' + upStreamFieldName, '').__str__()

            # print('上游表名:' + upStreamTableName)
            # print('上游字段名:' + upStreamFieldName)

            upStreamStrList.append(fieldUrn(upStreamTableName, upStreamFieldName))

            # 用于检查上游血缘是否冲突
            upStreamsList.append(Upstream(dataset=datasetUrn(upStreamTableName), type=DatasetLineageType.TRANSFORMED))

    fineGrainedLineage = FineGrainedLineage(upstreamType=FineGrainedLineageUpstreamType.DATASET,
                                            upstreams=upStreamStrList,
                                            downstreamType=FineGrainedLineageDownstreamType.FIELD_SET,
                                            downstreams=downStreamStrList)

    fineGrainedLineageList.append(fineGrainedLineage)

fieldLineages = UpstreamLineage(
    upstreams=upStreamsList, fineGrainedLineages=fineGrainedLineageList
)

lineageMcp = MetadataChangeProposalWrapper(
    entityUrn=datasetUrn(targetTableName),  # 下游表名
    aspect=fieldLineages
)

# 调用datahub REST API
emitter = DatahubRestEmitter('http://datahub-gms:8080') # datahub gms server

# Emit metadata!
emitter.emit_mcp(lineageMcp)

执行py脚本

python3 sql-lineage-to-datahub.py target.sql

查看web界面

sqllineage解析sql列级血缘并提交到datahub

sqllineage解析sql列级血缘并提交到datahub

 

 

到了这里,关于sqllineage解析sql列级血缘并提交到datahub的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【vscode和gitee】如何更改VsCode的gitee远程库地址,并提交到新的仓库中

    如果已经执行1 ,此步骤可以忽略 如果远程分支名和本地分支名称相同,则可以只写本地分支

    2024年02月11日
    浏览(45)
  • Hive SQL血缘解析

    Druid可以直接获得所有的列 http://t.csdn.cn/mO4TX 利用Hive提供的LineageLogger与Execution Hooks机制做血缘 https://blog.csdn.net/qq_44831907/article/details/123033137 Apache Calcite gudusoft 解析方案 商业 https://blog.csdn.net/qq_31557939/article/details/126277212 6.github开源项目: https://github.com/Shkin1/hathor https://github

    2024年02月04日
    浏览(32)
  • 【SQL解析】- SQL血缘分析实现篇01

    在数据中台中,通常我们面对的是海量的基于数仓的ETL、取数、建模、业务调用等等的数据操作任务,面对错综复杂的调度依赖关系,当出现问题需要快速追溯数据链路、以及对热点资源的盘点治理,人工进行整理效率太低。所以目前一般的数据平台都会实现一个数据地图(

    2024年02月10日
    浏览(42)
  • 基于Python-sqlparse的SQL表血缘追踪解析实现

    目录 前言 一、主线任务 1.数据治理 2.血缘追踪 3.SQL表血缘 二、实现过程 1.目标效果 2.代码实现 1.功能函数识别 2.SQL标准格式  3.解析AST树 4.最终效果: 点关注,防走丢,如有纰漏之处,请留言指教,非常感谢 之前我在两篇SQLparse的开源库解析中就说过自己在寻找在python编程

    2024年02月02日
    浏览(31)
  • [设计] Doris血缘解析流程

    元数据是凌久中台重要功能模块,是数据治理的重要一环,元数据治理是一切数据治理的基础,主要分为元数据管理和表血缘管理;        元数据管理主要用来做数据地图、数据资产等;        血缘治理主要用来追查问题数据上游表来源,以及分析当前表的对下游标

    2023年04月09日
    浏览(38)
  • spark-sql字段血缘实现

    Apache Spark是一个开源的大数据处理框架,它提供了一种高效、易于使用的方式来处理大规模数据集。在Spark中,数据是通过DataFrame和Dataset的形式进行操作的,这些数据结构包含了一系列的字段(也称为列)。字段血缘是Spark中的一个关键概念,它帮助我们理解数据的来源和流

    2024年02月02日
    浏览(50)
  • 基于Druid的HiveSQL血缘解析

    目录 前言 一、Druid简介 二、Druid SQL Parser Parser AST Visitor 三、血缘功能实现 1.建表语句 1.直接Create+字段定义 2. Create table... as select..  2.插入 1.标准语法 2.高级语法(Multiple Inserts) 3.高级语法(Dynamic Partition Inserts) 点关注,防走丢,如有纰漏之处,请留言指教,非常感谢 之前开发

    2024年02月21日
    浏览(28)
  • pycharm新建分支并提送至GitHub

    当我们写的项目代码越来越多时,一个master分支无法满足需求了,这个时候就需要创建分支来管理代码。 创建分支可以快速的回滚到某个节点的版本,也可以多个开发者同时开发一个项目,当自己的功能完成后,提交到自己的分支上。 在分支上测试通过后,最后把代码合并

    2024年02月16日
    浏览(42)
  • 数据血缘管理

    作为数据分析师,充分认识业务数据流,可以大大提高数据分析效率。 1.什么是数据血缘管理? 数据血缘管理是元数据管理的重要组成部门,它描述数据的上下游关系。 按血缘对象来分,可分为系统级血缘、表级血缘、字段(列)级血缘。不管是结构化数据还是非结构化数

    2024年02月04日
    浏览(39)
  • DataHub调研&数据血缘

    回答: 不是 DataHub是由Linkedin开源的,官方喊出的口号为:The Metadata Platform for the Modern Data Stack - 为现代数据栈而生的元数据平台。官方网站A Metadata Platform for the Modern Data Stack | DataHub。目的就是为了解决多种多样数据生态系统的元数据管理问题,它提供元数据检索、数据发现

    2024年02月07日
    浏览(37)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包