如何轻松做数据治理？开源技术栈告诉你答案

标签： dev | 发表时间：2022-12-28 00:00 | 作者：
出处：https://itindex.net/relian

data-lineage

搭建一套数据治理体系耗时耗力，但或许我们没有必要从头开始搞自己的数据血缘项目。本文分享如何用开源、现代的 DataOps、ETL、Dashboard、元数据、数据血缘管理系统构建大数据治理基础设施。

元数据治理系统

元数据治理系统是一个提供了所有数据在哪、格式化方式、生成、转换、依赖、呈现和所属的 一站式视图。

元数据治理系统是所有数据仓库、数据库、表、仪表板、ETL 作业等的 目录接口（catalog），有了它，我们就不用在群里喊“大家好，我可以更改这个表的 schema 吗？”、 “请问谁知道我如何找到 table-view-foo-bar 的原始数据？”…一个成熟的数据治理方案中的元数据治理系统，对数据团队来说非常必要。

而 数据血缘则是元数据治理系统众多需要管理的元数据之一，例如，某些 Dashboard 是某一个 Table View 的下游，而这个 Table View 又是从另外两个上游表 JOIN 而来。显然，应该清晰地掌握、管理这些信息，去构建一个可信、可控的系统和数据质量控制体系。

数据治理的可行方案

数据治理方案设计

元数据和数据血缘本质上非常适合采用图数据建模、图数据库。因为数据治理涉及的典型查询便是面向图关系的查询，像“查找指定组件（即表）的所有 n 度（深度）的数据血缘”就是图查询语句 FIND ALL PATH跑起来的事。从日常大家在 论坛、微信群里讨论的查询和图建模来看，NebulaGraph 社区很多人在从零开始搭建数据血缘系统，而这些工作看起来大多是在重复造轮子，而且还是不容易造的轮子。

既然如此，前人种树后人乘凉，这里我决定搭建一个完备、端到端（不只有元数据管理）的数据系统，供大家参考解决数据血缘、数据治理问题。这个套数据系统会采用市面上优秀的开源项目，而图数据库这块还是采用大家的老朋友——NebulaGraph。希望对大家能有所启发，在此基础之上拥有一个相对完善的图模型，以及设计精巧、开箱即用的元数据治理系统。

下面，来看看元数据治理系统的轮子都需要哪些功能组件：

  • 元数据抽取
    • 这部分需要从不同的数据栈拉/推数据，像是从数据库、数仓、Dashboard，甚至是 ETL Pipeline 和应用、服务中搞数据。
  • 元数据存储
    • 可以存在数据库、图数据库里，甚至存成超大的 JSON manifest 文件都行
  • 元数据目录接口系统 Catalog
    • 提供 API / GUI 来读写元数据和数据血缘系统

下图是整个方案的简单示意图：

其中，上面的虚线框是元数据的来源与导入、下面的虚线框是元数据的存储与展示、发现。

diagram-of-ref-project

开源技术栈

下面，介绍下数据治理系统的每个部分。

数据库和数仓

为了处理和使用原始和中间数据，这里一定涉及至少一个数据库或者数仓。它可以是 Hive、Apache Delta、TiDB、Cassandra、MySQL 或 Postgres。

在这个参考项目中，我们选一个简单、流行的 Postgres。

✓ 数据仓库：Postgres

数据运维 DataOps

我们应该有某种 DataOps 的方案，让 Pipeline 和环境具有可重复性、可测试性和版本控制性。

在这里，我们使用了 GitLab 创建的 Meltano。

Meltano 是一个 just-work 的 DataOps 平台，它可以用巧妙且优雅的方式将 Singer作为 EL 和 dbt作为 T 连接起来。此外，它还连接到其他一些 dataInfra 实用程序，例如 Apache Superset 和 Apache Airflow 等。

至此，我们又纳入了一个成员：

✓ GitOps：Meltano https://gitlab.com/meltano/meltano

ETL 工具

上面我们提到过组合 Singer与 Meltano 将来自许多不同数据源的数据 E（提取）和 L（加载）数据目标，并使用 dbt作为 Transform 的平台。于是我们得到：

✓ EL：Singer
✓ T: dbt

数据可视化

在数据之上创建 Dashboard、图表和表格得到数据的洞察是很符合直觉的，类似大数据之上的 Excel 图标功能。

Apache Superset是我很喜欢的开源数据可视化项目，我准备用它来作为被治理管理的目标之一。同时，还会利用它实现可视化功能来完成元数据洞察。于是，

✓ Dashboard：Apache Superset

任务编排（DAG Job Orchestration）

在大多数情况下，我们的 DataOps 作业、任务会演变成需要编排系统的规模，我们可以用 Apache Airflow来负责这一块。

✓ DAG：Apache Airflow https://airflow.apache.org/

元数据治理

随着越来越多的组件和数据被引入数据基础设施，在数据库、表、数据建模（schema）、Dashboard、DAG（编排系统中的有向无环图）、应用与服务的各个生命周期阶段中都将存着海量的元数据，需要对它们的管理员和团队进行协同管理、连接和发现。

Linux Foundation Amundsen是解决该问题的最佳项目之一。 Amundsen 用图数据库为事实源（single source of truth）以加速多跳查询，Elasticsearch 为全文搜索引擎。它在顺滑地处理所有元数据及其血缘之余，还提供了优雅的 UI 和 API。 Amundsen 支持多种图数据库为后端，这里咱们用 NebulaGraph。

现在的技术栈：

✓ 数据发现：Linux Foundation Amundsen
✓ 全文搜索：Elasticsearch
✓ 图数据库：NebulaGraph

好的，所有组件都齐正了，开始组装它们吧。

环境搭建与各组件初识

本次实践的项目方案已开源，你可以访问 https://github.com/wey-gu/data-lineage-ref-solution来获得对应的代码。

整个实践过程，我遵循了尽量干净、鼓励共建的原则。项目预设在一个 unix-like 系统上运行，且联网和装有 Docker-Compose。

这里，我将在 Ubuntu 20.04 LTS X86_64 上运行它，当然在其他发行版或 Linux 版本上应该也没有问题。

运行一个数仓、数据库

首先，安装 Postgres 作为我们的数仓。

这个单行命令会创建一个使用 Docker 在后台运行的 Postgres，进程关闭之后容器不会残留而是被清理掉（因为参数 --rm）。

   docker run --rm --name postgres \
    -e POSTGRES_PASSWORD=lineage_ref \
    -e POSTGRES_USER=lineage_ref \
    -e POSTGRES_DB=warehouse -d \
    -p 5432:5432 postgres

我们可以用 Postgres CLI 或 GUI 客户端来验证命令是否执行成功。

DataOps 工具链部署

接下来，安装有机结合了 Singer 和 dbt 的 Meltano。

Meltano 帮助我们管理 ETL 工具（作为插件）及其所有配置和 pipeline。这些元信息位于 Meltano 配置及其系统数据库中，其中配置是基于文件的（可以使用 GitOps 管理），它的默认系统数据库是 SQLite。

安装 Meltano

使用 Meltano 的工作流是启动一个“meltano 项目”并开始将 E、L 和 T 添加到配置文件中。Meltano 项目的启动只需要一个 CLI 命令 meltano init yourprojectname。不过，在那之前，先用 Python 的包管理器 pip 或者 Docker 镜像安装 Meltano，像我示范的这样：

在 Python 虚拟环境中使用 pip 安装 Meltano：

   mkdir .venv
# example in a debian flavor Linux distro
sudo apt-get install python3-dev python3-pip python3-venv python3-wheel -y
python3 -m venv .venv/meltano
source .venv/meltano/bin/activate
python3 -m pip install wheel
python3 -m pip install meltano

# init a project
mkdir meltano_projects && cd meltano_projects
# replace <yourprojectname> with your own one
touch .env
meltano init <yourprojectname>

或者，用 Docker 容器安装 Meltano：

   docker pull meltano/meltano:latest
docker run --rm meltano/meltano --version

# init a project
mkdir meltano_projects && cd meltano_projects

# replace <yourprojectname> with your own one
touch .env
docker run --rm -v "$(pwd)":/projects \
             -w /projects --env-file .env \
             meltano/meltano init <yourprojectname>

除了知晓 meltano init之外，最好掌握 Meltano 部分命令，例如 meltano etl表示 ETL 的执行， meltano invoke <plugin>来调用插件命令。详细可以参考它的速查表 https://docs.meltano.com/reference/command-line-interface。

Meltano GUI 界面

Meltano 自带一个基于 Web 的 UI，执行 ui子命令就能启动它：

   meltano ui

它默认会跑在 http://localhost:5000上。

针对 Docker 的运行环境，在暴露 5000 端口的情况下运行容器即可。由于容器的默认命令已经是 meltano ui，所以 run的命令只需：

   docker run -v "$(pwd)":/project \
             -w /project \
             -p 5000:5000 \
             meltano/meltano

Meltano 项目示例

GitHub 用户 Pat Nadolny在开源项目 singer_dbt_jaffle中做了很好的示例。他采用 dbt 的 Meltano 示例数据集，利用 Airflow 编排 ETL 任务。

不只这样，他还有利用 Superset 的例子，见 jaffle_superset。

前人种树我们来吃果，按照 Pat Nadolny 的实践，我们可以这样地运行数据管道（pipeline）：

  • tap-CSV（Singer）从 CSV 文件中提取数据
  • target-postgres（Singer） 将数据加载到 Postgres
  • dbt将数据转换为聚合表或视图

注意，上面我们已经启动了 Postgres，可以跳过容器启动 Postgres 这步。

操作过程是：

   git clone https://github.com/pnadolny13/meltano_example_implementations.git
cd meltano_example_implementations/meltano_projects/singer_dbt_jaffle/

meltano install
touch .env
echo PG_PASSWORD="lineage_ref" >> .env
echo PG_USERNAME="lineage_ref" >> .env

# Extract and Load(with Singer)
meltano run tap-csv target-postgres

# Trasnform(with dbt)
meltano run dbt:run

# Generate dbt docs
meltano invoke dbt docs generate

# Serve generated dbt docs
meltano invoke dbt docs to serve

# Then visit http://localhost:8080

现在，我们可以连接到 Postgres 来查看加载和转换后的数据预览。如下所示，截图来自 VS Code 的 SQLTool。

payments 表里长这样子：

搭一个 BI Dashboard 系统

现在，我们的数据仓库有数据了。接下来，可以试着用下这些数据。

像仪表盘 Dashbaord 这样的 BI 工具能帮我们从数据中获得有用的洞察。使用可视化工具 Apache Superset 可以很容易地创建和管理这些基于数据源的 Dashboard 和各式各样的图表。

本章的重点不在于 Apache Superset 本身，所以，咱们还是复用 Pat Nadolny 的 jaffle_superset例子。

Bootstrap Meltano 和 Superset

创建一个安装了 Meltano 的 Python VENV：

   mkdir .venv
python3 -m venv .venv/meltano
source .venv/meltano/bin/activate
python3 -m pip install wheel
python3 -m pip install meltano

参考 Pat 的小抄，做一些细微的调整：

克隆 repo，进入 jaffle_superset项目：

   git clone https://github.com/pnadolny13/meltano_example_implementations.git
cd meltano_example_implementations/meltano_projects/jaffle_superset/

修改 meltano 配置文件，让 Superset 连接到我们创建的 Postgres：

   vim meltano_projects/jaffle_superset/meltano.yml

这里，我将主机名更改为“10.1.1.111”，这是我当前主机的 IP。如果你是 Windows 或者 macOS 上的 Docker Desktop，这里不要修改主机名，否则就要和我一样手动改成实际地址：

   --- a/meltano_projects/jaffle_superset/meltano.yml
+++ b/meltano_projects/jaffle_superset/meltano.yml
@@ -71,7 +71,7 @@ plugins:
               A list of database driver dependencies can be found here https://superset.apache.org/docs/databases/installing-database-drivers
     config:
       database_name: my_postgres
-      sqlalchemy_uri: postgresql+psycopg2://${PG_USERNAME}:${PG_PASSWORD}@host.docker.internal:${PG_PORT}/${PG_DATABASE}
+      sqlalchemy_uri: postgresql+psycopg2://${PG_USERNAME}:${PG_PASSWORD}@10.1.1.168:${PG_PORT}/${PG_DATABASE}
       tables:
       - model.my_meltano_project.customers
       - model.my_meltano_project.orders

添加 Postgres 登录的信息到 .env文件：

   echo PG_USERNAME=lineage_ref >> .env
echo PG_PASSWORD=lineage_ref >> .env

安装 Meltano 项目，运行 ETL 任务：

   meltano install
meltano run tap-csv target-postgres dbt:run

调用、启动 Superset，这里注意 ui不是 meltano 的内部命令，而是一个配置进去的自定义行为（user-defined action）：

   meltano invoke superset:ui

在另一个命令行终端执行自定义的命令 load_datasources：

   meltano invoke superset:load_datasources

通过浏览器访问 http://localhost:8088/就是 Superset 的图形界面了：

创建一个 Dashboard

现在，我们站在 Meltano、Postgres 的肩膀上，用 ETL 数据建一个 Dashboard 吧：

点击 + DASHBOARD，填写仪表盘名称，再先后点击 SAVE和 + CREATE A NEW CHART：

在新图表（Create a new chart）视图中，选择图表类型和数据集。在这里，我选择了 orders表作为数据源和 Pie Chart图表类型：

点击 CREATE NEW CHART后，在图表定义视图中选择 “status” 的 “Query” 为 “DIMENSIONS”，“COUNT(amount)” 为 “METRIC”。至此，咱们就可以看到每个订单状态分布的饼图了。

点击 SAVE，系统会询问应该将此图表添加到哪个 Dashboard。选择后，单击 SAVE & GO TO DASHBOARD。

在 Dashboard 中，我们可以看到所有的图表。这不，你可以看到我额外添加的、用来显示客户订单数量分布的图表：

点 ···能看到刷新率设置、下载渲染图等其他的功能。

现在，我们有一个简单但典型的 HomeLAB 数据技术栈了，并且所有东西都是开源的！

想象一下，我们在 CSV 中有 100 个数据集，在数据仓库中有 200 个表，并且有几个数据工程师在运行不同的项目，这些项目使用、生成不同的应用与服务、Dashbaord 和数据库。当有人想要查找、发现或者修改其中的一些表、数据集、Dashbaord 和管道，在沟通和工程方面可能都是非常不好管理的。

上面我们提到，这个示例项目的 主要功能是元数据发现系统。

元数据发现系统

现在，需要我们部署一个带有 NebulaGraph 和 Elasticsearch 的 Amundsen。有了 Amundsen，我们可以在一个地方发现和管理整个数据栈中的所有元数据。

Amundsen 主要有两个部分组成：

它的工作原理是：利用 Databuilder提取不同数据源的元数据，并将元数据持久化到 Metadata Service和 Search Service中，用户从 Frontend Service或 Metadata Service的 API 获取数据。

部署 Amundsen

元数据服务 Metadata Service

我们用 docker-compose 部署一个 Amundsen 集群。由于 Amundsen 对 NebulaGraph 后端的支持 pr#1817尚未合并，还不能用官方的代码。这里，先用我的 fork 版本。

先克隆包含所有子模块的 repo：

   git clone -b amundsen_nebula_graph --recursive [email protected]:wey-gu/amundsen.git
cd amundsen

启动所有目录服务 catalog services 及其后端存储：

   docker-compose -f docker-amundsen-nebula.yml up

由于这个 docker-compose 文件是供开发人员试玩、调试 Amundsen 用的，而不是给生产部署准备的，它在启动的时候会从代码库构建镜像，第一次跑的时候启动会慢一些。

部署好了之后，我们使用 Databuilder 将一些示例、虚构的数据加载存储里。

抓取元数据 Databuilder

Amundsen Databuilder 就像 Meltano 系统一样，只不过是用在元数据的上的 ETL ，它把元数据加载到 Metadata Service和 Search Service的后端存储：NebulaGraph 和 Elasticsearch 里。这里的 Databuilder 只是一个 Python 模块，所有的元数据 ETL 作业可以作为脚本运行，也可以用 Apache Airflow 等 DAG 平台进行编排。

安装 Amundsen Databuilder：

   cd databuilder
python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install wheel
python3 -m pip install -r requirements.txt
python3 setup.py install

调用这个示例数据构建器 ETL 脚本来把示例的虚拟数据导进去。

   python3 example/scripts/sample_data_loader_nebula.py
验证一下 Amundsen

在访问 Amundsen 之前，我们需要创建一个测试用户：

   # run a container with curl attached to amundsenfrontend
docker run -it --rm --net container:amundsenfrontend nicolaka/netshoot

# Create a user with id test_user_id
curl -X PUT -v http://amundsenmetadata:5002/user \
    -H "Content-Type: application/json" \
    --data \
    '{"user_id":"test_user_id","first_name":"test","last_name":"user", "email":"[email protected]"}'

exit

然后我们可以在 http://localhost:5000查看 UI 并尝试搜索 test，它应该会返回一些结果。

然后，可以单击并浏览在 sample_data_loader_nebula.py期间加载到 Amundsen 的那些示例元数据。

此外，我们还可以通过 NebulaGraph Studio 的地址 http://localhost:7001访问 NebulaGraph 里的这些数据。

下图显示了有关 Amundsen 组件的更多详细信息：

   ┌────────────────────────┐ ┌────────────────────────────────────────┐
       │ Frontend:5000          │ │ Metadata Sources                       │
       ├────────────────────────┤ │ ┌────────┐ ┌─────────┐ ┌─────────────┐ │
       │ Metaservice:5001       │ │ │        │ │         │ │             │ │
       │ ┌──────────────┐       │ │ │ Foo DB │ │ Bar App │ │ X Dashboard │ │
  ┌────┼─┤ Nebula Proxy │       │ │ │        │ │         │ │             │ │
  │    │ └──────────────┘       │ │ │        │ │         │ │             │ │
  │    ├────────────────────────┤ │ └────────┘ └─────┬───┘ └─────────────┘ │
┌─┼────┤ Search searvice:5002   │ │                  │                     │
│ │    └────────────────────────┘ └──────────────────┼─────────────────────┘
│ │    ┌─────────────────────────────────────────────┼───────────────────────┐
│ │    │                                             │                       │
│ │    │ Databuilder     ┌───────────────────────────┘                       │
│ │    │                 │                                                   │
│ │    │ ┌───────────────▼────────────────┐ ┌──────────────────────────────┐ │
│ │ ┌──┼─► Extractor of Sources           ├─► nebula_search_data_extractor │ │
│ │ │  │ └───────────────┬────────────────┘ └──────────────┬───────────────┘ │
│ │ │  │ ┌───────────────▼────────────────┐ ┌──────────────▼───────────────┐ │
│ │ │  │ │ Loader filesystem_csv_nebula   │ │ Loader Elastic FS loader     │ │
│ │ │  │ └───────────────┬────────────────┘ └──────────────┬───────────────┘ │
│ │ │  │ ┌───────────────▼────────────────┐ ┌──────────────▼───────────────┐ │
│ │ │  │ │ Publisher nebula_csv_publisher │ │ Publisher Elasticsearch      │ │
│ │ │  │ └───────────────┬────────────────┘ └──────────────┬───────────────┘ │
│ │ │  └─────────────────┼─────────────────────────────────┼─────────────────┘
│ │ └────────────────┐   │                                 │
│ │    ┌─────────────┼───►─────────────────────────┐ ┌─────▼─────┐
│ │    │ NebulaGraph │   │                         │ │           │
│ └────┼─────┬───────┴───┼───────────┐     ┌─────┐ │ │           │
│      │     │           │           │     │MetaD│ │ │           │
│      │ ┌───▼──┐    ┌───▼──┐    ┌───▼──┐  └─────┘ │ │           │
│ ┌────┼─►GraphD│    │GraphD│    │GraphD│          │ │           │
│ │    │ └──────┘    └──────┘    └──────┘  ┌─────┐ │ │           │
│ │    │ :9669                             │MetaD│ │ │  Elastic  │
│ │    │ ┌────────┐ ┌────────┐ ┌────────┐  └─────┘ │ │  Search   │
│ │    │ │        │ │        │ │        │          │ │  Cluster  │
│ │    │ │StorageD│ │StorageD│ │StorageD│  ┌─────┐ │ │  :9200    │
│ │    │ │        │ │        │ │        │  │MetaD│ │ │           │
│ │    │ └────────┘ └────────┘ └────────┘  └─────┘ │ │           │
│ │    ├───────────────────────────────────────────┤ │           │
│ └────┤ Nebula Studio:7001                        │ │           │
│      └───────────────────────────────────────────┘ └─────▲─────┘
└──────────────────────────────────────────────────────────┘

穿针引线：元数据发现

设置好基本环境后，让我们把所有东西穿起来。还记得我们有 ELT 一些数据到 PostgreSQL 吗？

那么，我们如何让 Amundsen 发现这些数据和 ETL 的元数据呢？

提取 Postgres 元数据

我们从数据源开始：首先是 Postgres。

我们为 Python3 安装 Postgres 客户端：

   sudo apt-get install libpq-dev
pip3 install Psycopg2

执行 Postgres 元数据 ETL

运行一个脚本来解析 Postgres 元数据：

   export CREDENTIALS_POSTGRES_USER=lineage_ref
export CREDENTIALS_POSTGRES_PASSWORD=lineage_ref
export CREDENTIALS_POSTGRES_DATABASE=warehouse

python3 example/scripts/sample_postgres_loader_nebula.py

我们看看把 Postgres 元数据加载到 NebulaGraph 的示例脚本的代码，非常简单直接：

   # part 1: PostgresMetadata --> CSV --> NebulaGraph
job = DefaultJob(
      conf=job_config,
      task=DefaultTask(
          extractor=PostgresMetadataExtractor(),
          loader=FsNebulaCSVLoader()),
      publisher=NebulaCsvPublisher())

...
# part 2: Metadata stored in NebulaGraph --> Elasticsearch
extractor = NebulaSearchDataExtractor()
task = SearchMetadatatoElasticasearchTask(extractor=extractor)

job = DefaultJob(conf=job_config, task=task)

第一个工作路径是： PostgresMetadata --> CSV --> NebulaGraph

第二个工作路径是： Metadata stored in NebulaGraph --> Elasticsearch

  • NebulaSearchDataExtractor用于获取存储在 NebulaGraph 中的元数据
  • SearchMetadatatoElasticasearchTask用于使 Elasticsearch 对元数据进行索引。

请注意，在生产环境中，我们可以在脚本中或使用 Apache Airflow 等编排平台触发这些作业。

验证 Postgres 中元数据的获取

搜索 payments或者直接访问 http://localhost:5000/table_detail/warehouse/postgres/public/payments，你可以看到我们 Postgres 的元数据，比如：

像上面的屏幕截图一样，我们可以轻松完成元数据管理操作，如：添加标签、所有者和描述。

提取 dbt 元数据

其实，我们也可以从 dbt本身提取元数据。

Amundsen DbtExtractor会解析 catalog.json或 manifest.json文件并将元数据加载到 Amundsen 存储，这里当然指的是 NebulaGraph 和 Elasticsearch。

在上面的 Meltano 章节中，我们已经使用 meltano invoke dbt docs generate生成了这个文件：

   14:23:15  Done.
14:23:15  Building catalog
14:23:15  Catalog written to /home/ubuntu/ref-data-lineage/meltano_example_implementations/meltano_projects/singer_dbt_jaffle/.meltano/transformers/dbt/target/catalog.json

dbt 元数据 ETL 的执行

我们试着解析示例 dbt 文件中的元数据吧：

   $ ls -l example/sample_data/dbt/
total 184
-rw-rw-r-- 1 w w   5320 May 15 07:17 catalog.json
-rw-rw-r-- 1 w w 177163 May 15 07:17 manifest.json

我写的这个示例的加载例子如下：

   python3 example/scripts/sample_dbt_loader_nebula.py

其中主要的代码如下：

   # part 1: dbt manifest --> CSV --> NebulaGraph
job = DefaultJob(
      conf=job_config,
      task=DefaultTask(
          extractor=DbtExtractor(),
          loader=FsNebulaCSVLoader()),
      publisher=NebulaCsvPublisher())

...
# part 2: Metadata stored in NebulaGraph --> Elasticsearch
extractor = NebulaSearchDataExtractor()
task = SearchMetadatatoElasticasearchTask(extractor=extractor)

job = DefaultJob(conf=job_config, task=task)

它和 Postgres 元数据 ETL 的唯一区别是 extractor=DbtExtractor()，它带有以下配置以获取有关 dbt 项目的以下信息：

  • 数据库名称
  • 目录_json
  • manifest_json
   job_config = ConfigFactory.from_dict({
  'extractor.dbt.database_name': database_name,
  'extractor.dbt.catalog_json': catalog_file_loc,  # File
  'extractor.dbt.manifest_json': json.dumps(manifest_data),  # JSON Dumped objecy
  'extractor.dbt.source_url': source_url})

验证 dbt 抓取结果

搜索 dbt_demo或者直接访问 http://localhost:5000/table_detail/dbt_demo/snowflake/public/raw_inventory_value，可以看到

这里给一个小提示，其实，我们可以选择启用 DEBUG log 级别去看已发送到 Elasticsearch 和 NebulaGraph 的内容。

   - logging.basicConfig(level=logging.INFO)
+ logging.basicConfig(level=logging.DEBUG)

或者，在 NebulaGraph Studio 中探索导入的数据：

先点击 Start with Vertices，并填写顶点 vid： snowflake://dbt_demo.public/fact_warehouse_inventory

我们可以看到顶点显示为粉红色的点。再让我们修改下 Expand/ ”拓展“选项：

  • 方向：双向
  • 步数：单向、三步

并双击顶点（点），它将双向拓展 3 步：

像截图展示的那般，在可视化之后的图数据库中，这些元数据可以很容易被查看、分析，并从中获得洞察。

而且，我们在 NebulaGraph Studio 中看到的同 Amundsen 元数据服务的数据模型相呼应：

最后，请记住我们曾利用 dbt 来转换 Meltano 中的一些数据，并且清单文件路径是 .meltano/transformers/dbt/target/catalog.json，你可以尝试创建一个数据构建器作业来导入它。

提取 Superset 中的元数据

Amundsen 的 Superset Extractor 可以获取

来，现在试试提取之前创建的 Superset Dashboard 的元数据。

Superset 元数据 ETL 的执行

下边执行的示例 Superset 提取脚本可以获取数据并将元数据加载到 NebulaGraph 和 Elasticsearch 中。

   python3 sample_superset_data_loader_nebula.py

如果我们将日志记录级别设置为 DEBUG，我们实际上可以看到这些中间的过程日志：

   # fetching metadata from superset
DEBUG:urllib3.connectionpool:http://localhost:8088 "POST /api/v1/security/login HTTP/1.1" 200 280
INFO:databuilder.task.task:Running a task
DEBUG:urllib3.connectionpool:Starting new HTTP connection (1): localhost:8088
DEBUG:urllib3.connectionpool:http://localhost:8088 "GET /api/v1/dashboard?q=(page_size:20,page:0,order_direction:desc) HTTP/1.1" 308 374
DEBUG:urllib3.connectionpool:http://localhost:8088 "GET /api/v1/dashboard/?q=(page_size:20,page:0,order_direction:desc) HTTP/1.1" 200 1058
...

# insert Dashboard

DEBUG:databuilder.publisher.nebula_csv_publisher:Query: INSERT VERTEX `Dashboard` (`dashboard_url`, `name`, published_tag, publisher_last_updated_epoch_ms) VALUES  "superset_dashboard://my_cluster.1/3":("http://localhost:8088/superset/dashboard/3/","my_dashboard","unique_tag",timestamp());
...

# insert a DASHBOARD_WITH_TABLE relationship/edge

INFO:databuilder.publisher.nebula_csv_publisher:Importing data in edge files: ['/tmp/amundsen/dashboard/relationships/Dashboard_Table_DASHBOARD_WITH_TABLE.csv']
DEBUG:databuilder.publisher.nebula_csv_publisher:Query:
INSERT edge `DASHBOARD_WITH_TABLE` (`END_LABEL`, `START_LABEL`, published_tag, publisher_last_updated_epoch_ms) VALUES "superset_dashboard://my_cluster.1/3"->"postgresql+psycopg2://my_cluster.warehouse/orders":("Table","Dashboard","unique_tag", timestamp()), "superset_dashboard://my_cluster.1/3"->"postgresql+psycopg2://my_cluster.warehouse/customers":("Table","Dashboard","unique_tag", timestamp());

验证 Superset Dashboard 元数据

通过在 Amundsen 中搜索它，我们现在可以获得 Dashboard 信息。

我们也可以从 NebulaGraph Studio 进行验证。

注：可以参阅 Dashboard 抓取指南中的 Amundsen Dashboard 图建模：

用 Superset 预览数据

Superset 可以用来预览表格数据，文档可以参考 https://www.amundsen.io/amundsen/frontend/docs/configuration/#preview-client，其中 /superset/sql_json/的 API 被 Amundsen Frontend Service调用，取得预览信息。

开启数据血缘信息

默认情况下，数据血缘是关闭的，我们可以通过以下方式启用它：

第一步， cd到 Amundsen 代码仓库下，这也是我们运行 docker-compose -f docker-amundsen-nebula.yml up命令的地方：

   cd amundsen

第二步，修改 frontend 下的 TypeScript 配置

   --- a/frontend/amundsen_application/static/js/config/config-default.ts
+++ b/frontend/amundsen_application/static/js/config/config-default.ts
   tableLineage: {
-    inAppListEnabled: false,
-    inAppPageEnabled: false,
+    inAppListEnabled: true,
+    inAppPageEnabled: true,
     externalEnabled: false,
     iconPath: 'PATH_TO_ICON',
     isBeta: false,

第三步，重新构建 Docker 镜像，其中将重建前端图像。

   docker-compose -f docker-amundsen-nebula.yml build

第四步，重新运行 up -d以确保前端用新的配置：

   docker-compose -f docker-amundsen-nebula.yml up -d

结果大概长这样子：

   $ docker-compose -f docker-amundsen-nebula.yml up -d
...
Recreating amundsenfrontend           ... done

之后，我们可以访问 http://localhost:5000/lineage/table/gold/hive/test_schema/test_table1看到 Lineage （beta）血缘按钮已经显示出来了：

我们可以点击 Downstream查看该表的下游资源：

或者点击血缘按钮查看血缘的图表式：

也有用于血缘查询的 API。

下面这个例子中，我们用 cURL 调用下这个 API：

   docker run -it --rm --net container:amundsenfrontend nicolaka/netshoot

curl "http://amundsenmetadata:5002/table/snowflake://dbt_demo.public/raw_inventory_value/lineage?depth=3&direction=both"

上面的 API 调用是查询上游和下游方向的 linage，表 snowflake://dbt_demo.public/raw_inventory_value的深度为 3。

结果应该是这样的：

   {
    "depth": 3,
    "downstream_entities": [
        {
            "level": 2,
            "usage": 0,
            "key": "snowflake://dbt_demo.public/fact_daily_expenses",
            "parent": "snowflake://dbt_demo.public/fact_warehouse_inventory",
            "badges": [],
            "source": "snowflake"
        },
        {
            "level": 1,
            "usage": 0,
            "key": "snowflake://dbt_demo.public/fact_warehouse_inventory",
            "parent": "snowflake://dbt_demo.public/raw_inventory_value",
            "badges": [],
            "source": "snowflake"
        }
    ],
    "key": "snowflake://dbt_demo.public/raw_inventory_value",
    "direction": "both",
    "upstream_entities": []
}

实际上，这个血缘数据就是在我们的 dbtExtractor执行期间提取和加载的，其中 extractor .dbt.{DbtExtractor.EXTRACT_LINEAGE}默认为 true，因此，创建了血缘元数据并将其加载到了 Amundsen。

在 NebulaGraph 中洞察血缘

使用图数据库作为元数据存储的两个优点是：

图查询本身是一个灵活的 DSL for lineage API，例如，这个查询帮助我们执行 Amundsen 元数据 API 的等价的查询：

   MATCH p=(t:`Table`) -[:`HAS_UPSTREAM`|:`HAS_DOWNSTREAM` *1..3]->(x)
WHERE id(t) == "snowflake://dbt_demo.public/raw_inventory_value" RETURN p

来，在 NebulaGraph Studio 或者 Explorer 的控制台中查询下：

​渲染下这个结果：

提取数据血缘

这些血缘信息是需要我们明确指定、获取的，获取的方式可以是自己写 Extractor，也可以是用已有的方式。比如：dbt 的 Extractor 和 Open Lineage 项目的 Amundsen Extractor。

通过 dbt

这个在刚才已经展示过了，dbt 的 Extractor 会从表级别获取血缘同其他 dbt 中产生的元数据信息一起被拿到。

通过 Open Lineage

Amundsen 中的另一个开箱即用的血缘 Extractor 是 OpenLineageTableLineageExtractor。

Open Lineage是一个开放的框架，可以将不同来源的血统数据收集到一个地方，它可以将血统信息输出为 JSON 文件，参见文档 https://www.amundsen.io/amundsen/databuilder/#openlineagetablelineageextractor。

下边是它的 Amundsen Databuilder 例子：

   dict_config = {
    # ...
    f'extractor.openlineage_tablelineage.{OpenLineageTableLineageExtractor.CLUSTER_NAME}': 'datalab',
    f'extractor.openlineage_tablelineage.{OpenLineageTableLineageExtractor.OL_DATASET_NAMESPACE_OVERRIDE}': 'hive_table',
    f'extractor.openlineage_tablelineage.{OpenLineageTableLineageExtractor.TABLE_LINEAGE_FILE_LOCATION}': 'input_dir/openlineage_nd.json',
}
...

task = DefaultTask(
    extractor=OpenLineageTableLineageExtractor(),
    loader=FsNebulaCSVLoader())

回顾

整套元数据治理/发现的方案思路如下：

  • 将整个数据技术栈中的组件作为元数据源（从任何数据库、数仓，到 dbt、Airflow、Openlineage、Superset 等各级项目）
  • 使用 Databuilder（作为脚本或 DAG）运行元数据 ETL，以使用 NebulaGraph 和 Elasticsearch 存储和索引
  • 从前端 UI（使用 Superset 预览）或 API 去使用、消费、管理和发现元数据
  • 通过查询和 UI 对 NebulaGraph，我们可以获得更多的可能性、灵活性和数据、血缘的洞察

涉及到的开源

此参考项目中使用的所有项目都按字典顺序在下面列出。

  • Amundsen
  • Apache Airflow
  • Apache Superset
  • dbt
  • Elasticsearch
  • meltano
  • NebulaGraph
  • Open Lineage
  • Singer

谢谢你读完本文(///▽///)

如果你对本文感兴趣，现在可以用用 NebulaGraph Cloud 来搭建自己的数据治理系统哟，快来节省大量的部署安装时间来搞定业务吧~ Nebula Graph 阿里云计算巢现 30 天免费使用中， 点击链接来用用图数据库吧~

想看源码的小伙伴可以前往 GitHub 阅读、使用、(^з^)-☆ star 它 → GitHub；和其他的 NebulaGraph 用户一起交流图数据库技术和应用技能，留下 「你的名片」一起玩耍呢~

相关 [数据 治理 开源] 推荐：

如何轻松做数据治理？开源技术栈告诉你答案

- - IT瘾-dev
搭建一套数据治理体系耗时耗力，但或许我们没有必要从头开始搞自己的数据血缘项目. 本文分享如何用开源、现代的 DataOps、ETL、Dashboard、元数据、数据血缘管理系统构建大数据治理基础设施. 元数据治理系统是一个提供了所有数据在哪、格式化方式、生成、转换、依赖、呈现和所属的 一站式视图.

数据治理(Data Governance)

- - ITeye博客
数据治理是指从使用零散数据变为使用统一主数据、从具有很少或没有组织和流程治理到企业范围内的综合数据治理、从尝试处理主数据混乱状况到主数据井井有条的一个过程. 数据治理其实是一种体系，是一个关注于信息系统执行层面的体系，这一体系的目的是整合IT与业务部门的知识和意见，通过一个类似于监督委员会或项目小组的虚拟组织对企业的信息化建设进行全方位的监管，这一组织的基础是企业高层的授权和业务部门与IT部门的建设性合作.

如何成功实现数据治理

- - IT瘾-bigdata
如果你处理过大量数据，你也许听说过 “数据治理”一词，你可能会想， 它是什么. 简单来说， 数据治理就是处理数据的策略——如何存储、访问、验证、保护和使用数据. 数据治理包括制定获取方案：谁能访问、使用和共享你的数据. 这些问题正变得越来越重要， 因为企业依靠收集、存储和分析大量数据，来达成业务目标.

数据治理理论 + 实践

- - IT瘾-dev
数据治理无论是在数仓建设过程中还是数仓建设完成之后都是及其重要的，是数据部门基础建设的必经之路，是降本提效，形成企业数据资产的关键一环. 数据质量管理（Data Quality Management），是指对数据从计划、获取、存储、共享、维护、应用、消亡生命周期的每个阶段里可能引发的各类数据质量问题，进行识别、度量、监控、预警等一系列管理活动，并通过改善和提高组织的管理水平使得数据质量获得进一步提高.

腾讯内部数据治理实践

- -
导读：本文主要介绍目前腾讯数据治理的所在阶段和实践经验，以及基于目前的经验所沉淀的数据治理平台：WeData. 今天的介绍会围绕下面三方面展开：. WeData 数据治理平台能力. 分享嘉宾｜王浩仙 腾讯云 技术产品. 首先和大家分享腾讯在数据治理上所面临的挑战. 在数据治理的过程中会遇到很多问题，我们简单分成三类： 管理类挑战：数据信息分散在不同的业务部门的业务库中，数据上报也分散在不同的位置，难以对这些分散的数据进行统一的管理.

什么是数据治理？ | IBM

- -
数据治理通过不同的策略和标准提高组织数据的可用性、质量和安全性. 这些流程确定数据所有者、数据安全措施和数据的预期用途. 总体而言，数据治理的目标是维护安全且易于访问的高质量数据，以获取更深入的业务洞察. 大数据和数字化转型工作是数据治理计划的主要推动力量. 随着来自物联网 (IoT) 技术等新数据源的数据量的增加，组织需要重新考虑其数据管理实践，以扩展其商业智能.

建设大数据平台，从“治理”数据谈起 - CSDN极客头条

- -
一、大数据时代还需要数据治理吗. 二、如何面向用户开展大数据治理. 三、面向用户的自服务大数据治理架构. 一、大数据时代还需要数据治理吗. 数据平台发展过程中随处可见的数据问题. 大数据不是凭空而来，1981年第一个数据仓库诞生，到现在已经有了近40年的历史，相对数据仓库来说我还是个年轻人. 而国内企业数据平台的建设大概从90年代末就开始了，从第一代架构出现到现在已经经历了近20年的时间.

谈数据：微服务环境下，数据如何治理？ - 墨天轮

- -
前段时间，我的一个小伙伴跳槽到了某大型国有企业，刚到公司不久，老板给交给他一个重要项目——公司的数据中台规划. 老板交代：“要搞一个数据中台架构，涵盖数据资产管理、数据治理、数据分析等，同时这个数据中台，要体现去中心化，甚至无中心化的理念”. 我这哥们儿有过多年的数仓架构经验，并参考了业界主流的数据中台架构，很快就“照猫画虎”的搞了一个数据中台架构图出来.

上海数据治理的经验探索

- -
近年来，信息网络迅猛发展，已渗透到人类生产生活、社会经济运行的方方面面，构建了一个以网络为支撑的发展大环境. 随着网络应用的不断深化，大数据应用的影响与日俱增，国家政治、经济中绝大部分的活动都与数据的创造、采集、流通和使用有关，国家的核心竞争力也部分体现为拥有数据的规模、活性及分析、运用数据的能力.

数据治理项目失败，90%都被这样搞垮了

- -
数字化时代，数据作为新的生产要素受到了各界前所未有的重视. 随着数据越来越多，怎么管好、用好数据，让数据发挥价值，成为了很多企业的一个难题，而且还是一个必选题. 有效的数据治理可以确保企业数据全面、一致、可信，从而全面释放数据的价值，提高业务流程效率、提升业务增长的机会，驱动企业数字化转型. 这听起来很简单，但事实上数据治理对每个企业都是一项很大的挑战.