Openmetadata介绍

开源地址

在github的开源地址: Openmetadata

什么是Openmetadata?

Openmetadata 是一个用于数据发现、数据可观测性和数据治理的统一元数据平台，由中央元数据存储库、深入的列级沿袭和无缝团队协作提供支持。它是发展最快的开源项目之一，拥有充满活力的社区，并被各种垂直行业的各种公司采用。OpenMetadata 基于开放元数据标准和 API，支持各种数据服务的连接器，支持端到端元数据管理，让您可以自由释放数据资产的价值。

Openmetadata 由四个主要组件组成：

• 元数据架构：这些是基于常见抽象和类型的元数据的核心定义和词汇。它们还允许自定义扩展和属性，以适应不同的用例和域。

• 元数据存储：这是用于存储和管理元数据图的中央存储库，它以统一的方式连接数据资产、用户和工具生成的元数据。

• 元数据 API：这些是用于生成和使用元数据的接口，构建在元数据架构之上。它们支持将用户界面和工具、系统和服务与元数据存储无缝集成。

• Ingestion Framework：是一个可插拔的框架，用于将元数据从各种来源和工具摄取到元数据存储。它支持大约 75+ 个连接器，用于数据仓库、数据库、控制面板服务、消息传递服务、管道服务等。

Openmetadata的主要特性

• 数据发现：使用各种策略（例如关键字搜索、数据关联和高级查询）在一个位置查找和探索所有数据资产。您可以跨表、主题、控制面板、管道和服务进行搜索。

• 数据协作：与其他用户和团队就数据资产进行沟通、交谈和合作。您可以获取事件通知、发送提醒、添加公告、创建任务以及使用对话线程。

• 数据质量和分析器：使用无代码测量和监控质量，以建立对数据的信任。您可以定义和运行数据质量测试，将它们分组到测试套件中，并在交互式控制面板中查看结果。通过强大的协作，让数据质量成为您组织的共同责任。

• 数据管理：在整个组织中实施数据策略和标准。您可以定义数据域和数据产品，分配所有者和利益相关者，并使用标签和术语对数据资产进行分类。使用强大的自动化功能对数据进行自动分类。

• 数据洞察和 KPI：使用报告和平台分析来了解您组织的数据表现如何。Data Insights 提供所有关键指标的单一窗格视图，以最好地反映数据的状态。在 OpenMetadata 中定义关键绩效指标 （KPI） 并设定目标，以努力实现更好的文档、所有权和分层。可以根据要按指定计划接收的 KPI 设置警报。

• 数据血缘：端到端跟踪和可视化数据资产的来源和转换。您可以使用无代码编辑器手动查看列级世系、筛选查询和编辑世系。

• 数据文档：使用富文本、图像和链接记录您的数据资产和元数据实体。您还可以添加注释和注释，并生成数据字典和数据目录。

• 数据可观测性：监控数据资产和管道的运行状况和性能。您可以查看数据新鲜度、数据量、数据质量和数据延迟等指标。您还可以针对任何异常或故障设置警报和通知。

• 数据安全：使用各种身份验证和授权机制保护您的数据和元数据。您可以与不同的身份提供商集成以实现单点登录，并定义用于访问控制的角色和策略。

• Webhook：使用 Webhook 与外部应用程序和服务集成。您可以注册 URL 以接收元数据事件通知，并与 Slack、Microsoft Teams 和 Google Chat 集成。

• 连接器：使用连接器从各种来源和工具中提取元数据。OpenMetadata 支持大约 75+ 个连接器，用于数据仓库、数据库、控制面板服务、消息传递服务、管道服务等。

Openmetadata的工作原理

Openmetadata 的工作原理是将元数据从各种来源和工具摄取到中央存储库，然后使用 API 和 UI 进行管理和查询。以下是 Openmetadata 工作流程的详细说明：

• 首先是在数据源中捕获元数据，并将其摄取到 Openmetadata 的中央存储库。这可以通过使用连接器来完成，这些连接器支持各种数据服务，如GaussDB，MySQL，PostgreSQL等。

• 然后是在Ingestion Framework中定义摄取管道，并将其配置为从数据源中提取元数据并将其存储在中央存储库中。

• 接下来，使用元数据 API 和 UI 管理和查询存储在中央存储库中的元数据。这包括搜索、过滤和可视化数据资产以及执行数据质量测试等操作。

Openmetadata的连接器gaussdb开发步骤

上图介绍了新增GaussDB连接器的大概步骤。

GaussDB 介绍

什么是云数据库GaussDB

GaussDB是华为自主创新研发的分布式关系型数据库。该产品支持分布式事务，同城跨AZ部署，数据0丢失，支持1000+的扩展能力，PB级海量存储。同时拥有云上高可用，高可靠，高安全，弹性伸缩，一键部署，快速备份恢复，监控告警等关键能力，能为企业提供功能全面，稳定可靠，扩展性强，性能优越的企业级数据库服务。

在华为云官方可以访问 GaussDB 的帮助文档

整体架构

GaussDB分布式版形态整体架构

GaussDB分布式版形态整体架构如下：

图1 GaussDB分布式版形态整体架构图

• Coordinator Node：协调节点CN，负责接收来自应用的访问请求，并向客户端返回执行结果；负责分解任务，并调度任务分片在各DN上并行执行。

• GTM：全局事务管理器（Global Transaction Manager），负责生成和维护全局事务ID、事务快照、时间戳、Sequence信息等全局唯一的信息。

• Data Node：数据节点DN，负责存储业务数据、执行数据查询任务以及向CN返回执行结果。

GaussDB 集中式形态整体架构

GaussDB 集中式形态整体架构如下：

图2 GaussDB集中式形态整体架构图

• ETCD：分布式键值存储系统（Editable Text Configuration Daemon）。用于共享配置和服务发现（服务注册和查找）。

• CMS：集群管理模块（Cluster Manager）。管理和监控分布式系统中各个功能单元和物理资源的运行情况，确保整个系统的稳定运行。

• Data Node：数据节点DN，负责存储业务数据、执行数据查询任务以及返回执行结果。

产品优势

• 高安全

GaussDB拥有TOP级的商业数据库安全特性，如下所示，能够满足政企和金融级客户的核心安全诉求。 数据动态脱敏，行级访问控制，密态计算。

• 健全的工具与服务化能力

GaussDB已经拥有华为云，商用服务化部署能力，同时支持DAS、DRS等生态工具。有效保障用户开发、运维、优化、监控、迁移等日常工作需要。

• 全栈自研

GaussDB基于鲲鹏生态，是当前国内唯一能够做到全栈自主可控的国产品牌。同时GaussDB能够基于硬件优势在底层不断进行优化，提升产品综合性能。

• 开源生态

GaussDB已经支持开源社区，并提供集中式版本下载。

项目开发心得体会

在这篇文章中，我将从项目需求分析、代码开发以及Demo实现三个方面，分享我在适配GaussDB的OpenMetadata开源项目中的心得与体会。

项目需求分析

由于我已经非常熟悉GaussDB，并且接触多年，因此对这个项目充满信心。然而，在深入分析需求后，我发现OpenMetadata的复杂度超出了我的预期。项目涉及的开发语言不仅包括Python、Java，还包括TypeScript等，而项目本身包含多个模块。

为了更好地理解项目，我开始认真阅读OpenMetadata的文档，并梳理整体架构。幸运的是，OpenMetadata的文档比较详细，很快就理清了项目的框架和需求。

代码开发

在代码开发过程中，我参考了Postgres的代码实现。然而，我发现Postgres的内核与GaussDB相比更加复杂，且部分采集的系统表在GaussDB中并不存在。为了解决这一问题，我仔细对比了GaussDB与Postgres之间的差异，发现差异后，开发工作变得更加顺利。

此外，GaussDB对Python的支持有专有的驱动，不能使用开源的psycopg2。经过反复调试和测试，最终完成了代码的开发。

Demo实现

在设计Demo实现方案时，我进行了多方面的思考。由于目标是实现元数据采集，因此Demo应尽可能包含不同类型的对象。我在撰写Demo的SQL时，考虑了表、视图、存储过程、函数等多种对象类型。

为实现Demo，我首先对业务场景进行了分析，并绘制了整体部署架构图，最终基于CCE进行了部署。接下来，我完成了Demo的开发工作，首先在虚拟机ECS上实现，然后进行容器化改造。经过一步步的改进，最终成功完成了Demo的开发、部署与验证。

在华为云部署项目

Demo介绍

开发任务中对Demo的要求如下：

• Openmetadata 安装成功后使用GaussDB 存储其采集到的元数据信息。

• 可以通过 openmetadata WebUI 管理存储在GaussDB内的元数据信息。

经过分析，采用在GaussDB中创建样例数据库，并添加一些表，视图，存储过程等对象。添加测试数据，然后通过openmetadata采集元数据信息。

Demo部署架构

下图是Demo的部署架构，基于CCE容器化部署：

1.终端层

支持不同的类型终端访问Web UI，例如浏览器、手机等

2.网关层

通过ELB进行负载均衡代理

3.中间层

在CCE中部署3个无状态服务：

• openmetadata-server: 负责提供openmetadata的Web UI

• openmetadata-ingestion: 负责采集元数据信息

• execute-migrate-all: 负责执行数据库迁移脚本

在CCE中部署2个有状态服务：

• openmetadata-mysql: 负责存储openmetadata的元数据信息

• openmetadata-elasticsearch: 负责存储openmetadata的元数据信息索引

4.数据库层

在GaussDB中创建样例数据库，并添加一些表、视图、存储过程等对象。

步骤一：云资源购买与配置

Demo部署主要依赖的云资源如下：

• VPC: 虚拟私有云，实现隔离

• ECS：制作镜像

• ELB：负载均衡和代理

• GaussDB：数据库加工原始数据

• CCE：容器化部署

• SWR: 存放容器镜像

• NAT网关: 容器中访问公网

购买ECS并克隆项目

ECS配置如下：

• 购买数量：1台

• 规格：鲲鹏通用计算增强型 | kc1.2xlarge.4 | 8vCPUs | 32GiB

• 镜像: Huawei Cloud EulerOS 2.0 标准版 64位 ARM版

• 登录凭证：密码

• 系统盘: 通用型SSD, 40GiB

具体购买操作可参考 快速购买和使用Linux ECS

购买后，待服务器启动，登录服务器按下面的操作继续。

安装Docker

参考 基于EulerOS部署Docker

安装Git

参考 基于EulerOS配置GitHub

克隆项目并且制作镜像

# 克隆项目

cd /opt/cyl

git clone git@github.com:pangpang20/OpenMetadata.git

cd OpenMetadata
# 创建虚拟环境

python3 -m venv .venv

source .venv/bin/activate
# 环境要求：

Docker 20 or higher

Java JDK 17

Antlr 4.9.2 - sudo make install_antlr_cli

JQ - brew install jq (osx) apt-get install jq (Ubuntu)

Maven 3.5.x or higher - (with Java JDK 11)

Python 3.8 or 3.9

Node 18.x

Yarn ^1.22.0

Rpm (Optional, only to run RPM profile with maven)
# 编译镜像

sh docker/run_local_docker.sh

上传镜像到SWR

进入容器镜像服务 SWR，点击 组织管理 ,点击 创建组织

点击总览，点击右上角 登录指令，复制到ECS中执行

上传镜像

sudo docker tag {镜像名称}:{版本名称} swr.cn-south-1.myhuaweicloud.com/{组织名称}/{镜像名称}:{版本名称}
sudo docker push swr.cn-south-1.myhuaweicloud.com/{组织名称}/{镜像名称}:{版本名称}

上传后在我的镜像中可以查看

购买GaussDB并创建用户和数据库

GaussDB配置如下：

• 数据库引擎: GaussDB

• 数据库引擎版本:V2.0-8.201

• 内核引擎版本:505.2.0

• 实例类型: 集中式版

• 部署形态: 1主2备

• 性能规格: 通用型（1:4）| 4 vCPUs | 16 GB

• 存储空间: 40 GB

• 数据库端口: 默认端口8000

具体购买操作可参考 购买并通过界面 化工具 DAS连接GaussDB实例（推荐）

购买后，待数据库实例启动，登录DAS创建数据库和创建用户。

🔔 注意： 这里需要记录EIP或内网IP，数据库名，用户名，密码，后面需要。

样例sql

在DAS中执行以下SQL，创建数据库和表等对象。

创建数据库：metadb

-- 1. 创建表

DROP TABLE IF EXISTS products;
DROP TABLE IF EXISTS categories;
DROP TABLE IF EXISTS sales;
DROP TABLE IF EXISTS customers;
DROP TABLE IF EXISTS inventory;

-- 商品类别表

CREATE TABLE categories (
  category_id SERIAL PRIMARY KEY, 
  name VARCHAR(255) NOT NULL
);

-- 添加描述

COMMENT ON TABLE categories IS 'Table storing product categories';
COMMENT ON COLUMN categories.category_id IS 'The unique identifier for each category';
COMMENT ON COLUMN categories.name IS 'The name of the product category';

-- 商品表

CREATE TABLE products (
  product_id SERIAL PRIMARY KEY, 
  name VARCHAR(255) NOT NULL, 
  description TEXT, 
  price DECIMAL(10, 2) NOT NULL, 
  category_id INT, 
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 添加描述

COMMENT ON TABLE products IS 'Table storing product information';
COMMENT ON COLUMN products.product_id IS 'The unique identifier for each product';
COMMENT ON COLUMN products.name IS 'The name of the product';
COMMENT ON COLUMN products.description IS 'A description of the product';
COMMENT ON COLUMN products.price IS 'The price of the product';
COMMENT ON COLUMN products.category_id IS 'The category ID of the product';
COMMENT ON COLUMN products.created_at IS 'The creation timestamp of the product record';

-- 销售记录表

CREATE TABLE sales (
  sale_id SERIAL PRIMARY KEY, 
  product_id INT NOT NULL, 
  customer_id INT NOT NULL, 
  sale_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 
  quantity INT NOT NULL, 
  total DECIMAL(10, 2) NOT NULL
);

-- 添加描述

COMMENT ON TABLE sales IS 'Table storing sales transactions';
COMMENT ON COLUMN sales.sale_id IS 'The unique identifier for each sale transaction';
COMMENT ON COLUMN sales.product_id IS 'The ID of the product being sold';
COMMENT ON COLUMN sales.customer_id IS 'The ID of the customer making the purchase';
COMMENT ON COLUMN sales.sale_date IS 'The date and time the sale occurred';
COMMENT ON COLUMN sales.quantity IS 'The quantity of the product sold';
COMMENT ON COLUMN sales.total IS 'The total sale amount for the transaction';

-- 客户信息表

CREATE TABLE customers (
  customer_id SERIAL PRIMARY KEY, 
  name VARCHAR(255) NOT NULL, 
  email VARCHAR(255) UNIQUE, 
  phone VARCHAR(15)
);

-- 添加描述

COMMENT ON TABLE customers IS 'Table storing customer information';
COMMENT ON COLUMN customers.customer_id IS 'The unique identifier for each customer';
COMMENT ON COLUMN customers.name IS 'The name of the customer';
COMMENT ON COLUMN customers.email IS 'The email address of the customer';
COMMENT ON COLUMN customers.phone IS 'The phone number of the customer';

-- 库存表

CREATE TABLE inventory (
  product_id INT PRIMARY KEY, 
  stock_quantity INT NOT NULL
);

-- 添加描述

COMMENT ON TABLE inventory IS 'Table storing inventory data';
COMMENT ON COLUMN inventory.product_id IS 'The ID of the product in the inventory';
COMMENT ON COLUMN inventory.stock_quantity IS 'The available stock quantity of the product';

-- 2. 插入测试数据

-- 插入商品类别数据

INSERT INTO categories (name) VALUES
('Electronics'),
('Clothing'),
('Books'),
('Food');

-- 插入商品数据

INSERT INTO products (name, description, price, category_id) VALUES
('Laptop', 'High performance laptop', 999.99, 1),
('T-shirt', 'Cotton T-shirt', 19.99, 2),
('Novel', 'A gripping mystery novel', 12.99, 3),
('Apple', 'Fresh organic apples', 2.99, 4),
('Headphones', 'Noise-cancelling headphones', 149.99, 1);

-- 插入客户数据

INSERT INTO customers (name, email, phone) VALUES
('Alice', 'alice@example.com', '123-456-7890'),
('Bob', 'bob@example.com', '234-567-8901'),
('Charlie', 'charlie@example.com', '345-678-9012');

-- 插入库存数据

INSERT INTO inventory (product_id, stock_quantity) VALUES
(1, 50),
(2, 200),
(3, 100),
(4, 500),
(5, 30);

-- 插入销售记录数据

INSERT INTO sales (product_id, customer_id, quantity, total) VALUES
(1, 1, 2, 1999.98),
(2, 2, 3, 59.97),
(3, 3, 1, 12.99),
(4, 1, 10, 29.90),
(5, 2, 1, 149.99);

COMMIT;

-- 3. 创建存储过程

-- 创建存储过程：add_sale

CREATE OR REPLACE PROCEDURE add_sale(
  p_product_id INT,
  p_customer_id INT,
  p_quantity INT
)
AS
BEGIN
  DECLARE
    v_total DECIMAL(10, 2);
  BEGIN
    SELECT price * p_quantity INTO v_total
    FROM products
    WHERE product_id = p_product_id;

    -- 插入销售记录
    INSERT INTO sales (product_id, customer_id, quantity, total)
    VALUES (p_product_id, p_customer_id, p_quantity, v_total);

    -- 更新库存
    UPDATE inventory
    SET stock_quantity = stock_quantity - p_quantity
    WHERE product_id = p_product_id;
  END;
END;

-- 创建存储过程：add_product

CREATE OR REPLACE PROCEDURE add_product(
  p_product_name VARCHAR,
  p_price DECIMAL(10, 2),
  p_stock_quantity INT
)
AS
BEGIN
  -- 插入产品记录
  INSERT INTO products (name, price, stock_quantity)
  VALUES (p_product_name, p_price, p_stock_quantity);
END;

-- 创建存储过程：update_product_price

CREATE OR REPLACE PROCEDURE update_product_price(
  p_product_id INT,
  p_new_price DECIMAL(10, 2)
)
AS
BEGIN
  -- 更新产品价格
  UPDATE products
  SET price = p_new_price
  WHERE product_id = p_product_id;
END;

-- 创建存储过程：add_customer

CREATE OR REPLACE PROCEDURE add_customer(
  p_customer_name VARCHAR,
  p_email VARCHAR
)
AS
BEGIN
  -- 插入客户记录
  INSERT INTO customers (name, email)
  VALUES (p_customer_name, p_email);
END;

-- 创建存储过程：add_inventory

CREATE OR REPLACE PROCEDURE add_inventory(
  p_product_id INT,
  p_additional_quantity INT
)
AS
BEGIN
  -- 更新库存
  UPDATE inventory
  SET stock_quantity = stock_quantity + p_additional_quantity
  WHERE product_id = p_product_id;
END;

-- 创建函数：get_total_sales

CREATE OR REPLACE FUNCTION get_total_sales(p_product_id INT)
RETURNS DECIMAL AS $$
DECLARE
  v_total_sales DECIMAL(10, 2);
BEGIN
  -- 计算商品的总销售额
  SELECT SUM(total) INTO v_total_sales
  FROM sales
  WHERE product_id = p_product_id;

  -- 返回总销售额，如果没有销售记录则返回 0
  RETURN COALESCE(v_total_sales, 0);
END;
$$ LANGUAGE plpgsql;

-- 创建函数：get_customer_purchase_history

CREATE OR REPLACE FUNCTION get_customer_purchase_history(p_customer_id INT)
RETURNS TABLE(product_name VARCHAR, quantity INT, total DECIMAL) AS $$
BEGIN
  -- 返回客户购买的商品信息
  RETURN QUERY
  SELECT p.name, s.quantity, s.total
  FROM sales s
  JOIN products p ON s.product_id = p.product_id
  WHERE s.customer_id = p_customer_id;
END;
$$ LANGUAGE plpgsql;

-- 4. 创建视图

-- 商品销售详情

CREATE VIEW product_sales_view AS
SELECT p.name AS product_name,
       c.name AS category_name,
       SUM(s.quantity) AS total_quantity_sold,
       SUM(s.total) AS total_sales
FROM sales s
JOIN products p ON s.product_id = p.product_id
JOIN categories c ON p.category_id = c.category_id
GROUP BY p.name, c.name;

-- 客户购买历史

CREATE VIEW customer_purchase_history_view AS
SELECT c.name AS customer_name,
       p.name AS product_name,
       s.quantity,
       s.total,
       s.sale_date
FROM sales s
JOIN products p ON s.product_id = p.product_id
JOIN customers c ON s.customer_id = c.customer_id;

-- 查询商品销售详情：

SELECT * FROM product_sales_view;

-- 查询客户购买历史：

SELECT * FROM customer_purchase_history_view WHERE customer_name = 'Alice';

-- 获取商品的总销售额：

SELECT get_total_sales(1);

-- 获取客户的购买历史：

SELECT * FROM get_customer_purchase_history(1);

购买CCE和节点

CCE配置如下：

• 集群类型CCE: Turbo

• 容器网络模型云原生网络: 2.0

• 集群版本: v1.30

• 集群规模: 50 节点

• 集群 master 实例数: 3实例（高可用）

CCE集群创建后，创建节点，节点配置如下：

• 节点类型：弹性云服务器-虚拟机

• 节点规格：鲲鹏通用计算增强型 | kc2.2xlarge.4 | 8 vCPUs | 32 GiB

• 容器引擎：Docker

• 操作系统：Huawei Cloud EulerOS2.0

• 登录方式：密码

• 磁盘：默认

• 节点数量：3

具体购买操作可参考 在CCE集群中部署NGINX无状态工作负载

购买ELB

ELB配置如下：

• 实例类型：独享型

• 实例规格：弹性规格，应用型+网络型

• 所属VPC：和CCE在同一个VPC

• 弹性公网IP带宽：10 Mbit/s

具体购买操作可参考 实现单个Web应用的负载均衡

步骤二：部署Demo

可以参考 OpenMetadata/docker/development/docker-compose.yml 来配置相关参数

有状态的工作负载

在CCE中添加2个有状态的工作负载，具体如下：

openmetadata-mysql

负责存储openmetadata的元数据信息

openmetadata-elasticsearch

负责存储openmetadata的元数据信息索引

部署后如下：

无状态的工作负载

在CCE中添加3个无状态的工作负载，具体如下：

openmetadata-server

负责提供openmetadata的Web UI

openmetadata-ingestion

负责采集元数据信息

execute-migrate-all

负责执行数据库迁移脚本

部署后如下：

步骤三：访问UI

访问GaussDB

基于前面的demo SQL，我们在GaussDB中执行SQL语句，查看数据：

表对象：

视图对象：

存储过程对象：

接下来，我们通过openmetadata采集这些元数据信息。

访问airflow

在浏览器访问ELB的公网IP，访问airflow地址：

输入用户名和密码（admin，admin），进入airflow界面：

这时候还没有DAG任务，我们接下来通过在openmetadata-server创建DAG任务。

访问openmetadata-server

在浏览器访问ELB的公网IP，访问openmetadata-server地址：

输入用户名和密码（admin@open-metadata.org，admin），进入openmetadata-server界面：

按下图点击 Settings：

按下图点击 Services：

按下图点击 Databases：

按下图点击 Add new service：

按下图点击 Gaussdb, 然后点击Next： 

按下图填写相关信息，然后点击Next： 

按下图填写相关信息，然后点击Save： 

按下图点击 Add Ingestion:

按下图填写相关信息，然后点击Next:

配置手工执行，然后点击Add & Deploy：

在这里再添加两个ingestion:

一个是采集血缘信息：

一个是采集剖析信息：

添加完成后，这里可以看到3个ingestion：

• 先点击Type为metadata的ingestion，然后点击Run：

• 运行完成后，再点击Type为profiler的ingestion，然后点击Run：

• 运行完成后，再点击Type为lineage的ingestion，然后点击Run：

执行完成后，可以看到如下界面：

也可以在airflow中看到执行情况：

查看采集的元数据信息

点开左侧菜单，可以看到采集的元数据信息：

点开products表，可以看到采集的表信息：

表的样例数据：

表的剖析信息：

列的剖析信息：

如果是视图，还可以看到视图依赖的表信息，即数据血缘信息：

通过上面的查询，查看存储过程：

到此，GaussDB的元数据信息已经采集完成。