利用Apache Atlas进行数据资产盘点的实操方法与工具使用

Zhou Xin 2026-09-19 03:35

一、为什么需要做数据资产盘点

很多公司跑着几千张表、几百个任务，数据散落在各个集群里。问一句“咱们有哪些核心数据表？”很少有人能答全。数据资产盘点就是把这些散落的表、文件、指标、报表给梳理清楚，让所有人都知道数据在哪、谁在用、怎么来的。这活儿听着简单，做起来头大，因为元数据常常比数据本身还乱。

Apache Atlas就是专门解决这类问题的开源工具。它可以把各种数据源的元数据集中存起来，还能记录表和表之间的血缘关系。今天咱们就一起看看，怎么用Atlas做一次不那么痛苦的数据资产盘点。

二、Atlas能帮我们做什么

先得知道Atlas是个啥。你可以把它想象成一个超级档案室，专门保存“关于数据的数据”，也就是元数据。比如一张表叫什么、有哪些字段、字段类型是什么、属于哪个业务线、数据是从哪张表加工出来的，这些信息都能记录在Atlas里。

它最厉害的两个本事：一个是分类，给数据打标签，比如“个人敏感信息”“交易数据”“内部公开”，有了标签就能快速筛出需要重点关注的数据；另一个是血缘，能从一张表的上游一直追到下游，数据出问题的时候，顺着血缘就能定位到源头。

三、实操前的准备工作

3.1 跑起一个Atlas实例

咱们不聊太深奥的安装，简单说下思路。Atlas常见的启动方式有两种：自己搭Hadoop生态，或者用Docker方式。如果你只是做盘点练习，推荐用Docker，省时省力。假设你的Atlas已经通过http://localhost:21000可以访问了，那就可以开始干活。

3.2 技术栈说明

接下来的所有示例，统一使用Python 3加requests库，通过调用Atlas的REST API来完成操作。之所以选Python，是因为它在数据处理领域最常见，读起来也像大白话。

3.3 先给自己发一张“门禁卡”

Atlas API需要认证，默认用户名和密码通常是admin/admin。咱们写个公共函数，把认证信息带上。

# 技术栈：Python 3 + requests

import requests
import json

# Atlas服务地址，换成你自己的实际地址
ATLAS_BASE_URL = "http://localhost:21000/api/atlas/v2"

# 认证信息（默认账号密码）
AUTH = ("admin", "admin")

# 请求头，统一用JSON格式
HEADERS = {
    "Content-Type": "application/json"
}

def atlas_get(endpoint, params=None):
    """
    发送GET请求到Atlas
    :param endpoint: 接口路径，比如 /types/typedefs
    :param params: 查询参数，字典类型
    :return: 解析后的JSON对象
    """
    url = f"{ATLAS_BASE_URL}/{endpoint.lstrip('/')}"
    response = requests.get(url, auth=AUTH, headers=HEADERS, params=params)
    response.raise_for_status()  # 状态码不是200就抛异常
    return response.json()

def atlas_post(endpoint, body):
    """
    发送POST请求到Atlas
    :param endpoint: 接口路径
    :param body: 请求体，字典类型
    :return: 解析后的JSON对象
    """
    url = f"{ATLAS_BASE_URL}/{endpoint.lstrip('/')}"
    response = requests.post(url, auth=AUTH, headers=HEADERS, data=json.dumps(body))
    response.raise_for_status()
    return response.json()

有了这两个函数，后面所有的API调用都只需要写API路径和参数，省心很多。

四、把数据源信息喂给Atlas

盘点的第一步，是让Atlas认识你所拥有的数据。假设我们有两张Hive表：一张是订单明细表ods_order_detail，另一张是订单汇总表dws_order_summary。这两张表现在还没有登记在Atlas里，咱们通过API把它们创建成“数据实体”。

4.1 创建表类型

Atlas内置了很多类型，比如Hive表类型hive_table，字段类型hive_column。如果内置类型够用，就不需要自定义类型。这里我们直接用内置类型。

先创建一张表实体，注意传入attributes里的qualifiedName、name、owner这些属性。

# 技术栈：Python 3 + requests

def create_table(qualified_name, name, owner, description):
    """
    在Atlas中创建一个Hive表实体
    :param qualified_name: 唯一名称，通常为 库名.表名@集群名
    :param name: 表名
    :param owner: 负责人
    :param description: 描述信息
    :return: 创建的实体对象
    """
    table_entity = {
        "entity": {
            "typeName": "hive_table",  # 使用内置的hive_table类型
            "attributes": {
                "qualifiedName": qualified_name,
                "name": name,
                "owner": owner,
                "comment": description,  # comment字段存储表描述
                "createTime": 1699000000000,  # 用毫秒时间戳代替创建时间
                "lastAccessTime": 1699000000000
            }
        }
    }
    # 调用Atlas的实体创建接口
    result = atlas_post("/entity", table_entity)
    print(f"表 {name} 创建成功，GUID = {result['guidAssignments'][qualified_name]}")
    return result

# 创建两张示例表
create_table(
    "default.ods_order_detail@primary",
    "ods_order_detail",
    "数据平台组",
    "订单明细表，每天从业务库同步一次"
)

create_table(
    "default.dws_order_summary@primary",
    "dws_order_summary",
    "数据分析组",
    "订单汇总表，按天汇总订单数据"
)

运行这段代码，Atlas里就有这两张表了。接着咱们给表添加字段信息。字段在Atlas里是独立的实体，需要挂到表下面。看下面这段代码。

# 技术栈：Python 3 + requests

def add_column(table_qualified_name, column_name, data_type):
    """
    给某张表添加一个字段实体
    :param table_qualified_name: 所属表的qualifiedName
    :param column_name: 字段名
    :param data_type: 字段类型，比如string, int
    """
    column_entity = {
        "entity": {
            "typeName": "hive_column",
            "attributes": {
                "qualifiedName": f"{table_qualified_name}.{column_name}",
                "name": column_name,
                "type": data_type,
                "comment": f"{column_name}字段"
            }
        }
    }
    atlas_post("/entity", column_entity)
    print(f"字段 {column_name} 添加成功")

# 给 ods_order_detail 加几个核心字段
add_column("default.ods_order_detail@primary", "order_id", "string")
add_column("default.ods_order_detail@primary", "user_id", "string")
add_column("default.ods_order_detail@primary", "amount", "double")
add_column("default.ods_order_detail@primary", "create_time", "timestamp")

# 给 dws_order_summary 加汇总字段
add_column("default.dws_order_summary@primary", "date", "string")
add_column("default.dws_order_summary@primary", "total_amount", "double")
add_column("default.dws_order_summary@primary", "order_count", "int")

字段建好后，表的骨架就成了。但是注意，我们还没把表和字段关联起来。因为Atlas里每个字段实体都要通过table属性指定它属于哪张表。刚才的写法少了一步。正确的做法是在创建字段时，把所属表的引用传进去。咱们重新来一下，先删除刚才孤立的字段（这里省略删除代码），用正确的方式重新创建。

# 技术栈：Python 3 + requests

def add_column_with_table(table_qualified_name, column_name, data_type, table_guid):
    """
    给表添加字段，并建立与表的关联
    :param table_qualified_name: 表的qualifiedName
    :param column_name: 字段名
    :param data_type: 字段类型
    :param table_guid: 所属表的GUID
    """
    column_entity = {
        "entity": {
            "typeName": "hive_column",
            "attributes": {
                "qualifiedName": f"{table_qualified_name}.{column_name}",
                "name": column_name,
                "type": data_type,
                "comment": f"{column_name}字段"
            },
            "relationshipAttributes": {
                "table": {   # 指向所属表
                    "guid": table_guid,
                    "typeName": "hive_table"
                }
            }
        }
    }
    atlas_post("/entity", column_entity)
    print(f"字段 {column_name} 已关联到表 {table_qualified_name}")

# 假设下表GUID是通过创建表时拿到的，这里演示时用占位符
# 实际运行时需要替换为真实的GUID字符串
import uuid
table_guid_ods = str(uuid.uuid4())  # 这里仅为示例，真实GUID应从Atlas返回获取

add_column_with_table("default.ods_order_detail@primary", "order_id", "string", table_guid_ods)

你可能觉得麻烦，没关系，实际当中我们可以通过Atlas自带的Hive同步工具，一键把Hive所有表结构导入。咱们这里手动创建是为了让你更清楚背后的原理。

4.2 给数据打上分类标签

数据盘点的关键之一是识别敏感数据。比如订单表中包含用户ID、订单金额，这些属于敏感信息。Atlas支持用分类标签来标记。

创建分类之前，需要先定义分类类型。如果内置类型中没有你想要的，可以自己定义。下面演示定义一个“敏感数据”分类，然后给表打上标签。

# 技术栈：Python 3 + requests

def create_classification_type(name, description):
    """
    定义一个新的分类类型
    :param name: 分类类型名称
    :param description: 描述
    """
    classification_def = {
        "classificationDefs": [
            {
                "name": name,
                "description": description,
                "superTypes": [],   # 如果有父分类，可以填在这里
                "attributeDefs": [] # 分类上没有自定义属性，先留空
            }
        ]
    }
    atlas_post("/types/typedefs", classification_def)
    print(f"分类类型 {name} 创建成功")

# 创建“敏感数据”分类
create_classification_type("SensitiveData", "标识敏感数据，需要重点保护")

现在把分类标签挂到表实体上。Atlas里给实体打标签的接口是POST /entity/guid/{guid}/classification。

# 技术栈：Python 3 + requests

def add_classification_to_entity(entity_guid, classification_name):
    """
    给指定实体打上分类标签
    :param entity_guid: 实体GUID
    :param classification_name: 分类名称
    """
    classification_payload = {
        "classification": {
            "typeName": classification_name,
            "attributes": {
                # 这里如果分类有自定义属性，可以在这里填写
            }
        }
    }
    url = f"/entity/guid/{entity_guid}/classification"
    atlas_post(url, classification_payload)
    print(f"实体 {entity_guid} 已标记为 {classification_name}")

# 假设我们已经从创建表接口获得了表实体的GUID
# 实际应用时，这个GUID应该是真实返回的
# 比如 add_classification_to_entity("9d5f4c18-3b2e-4f7b-b212-aa19f3f2c123", "SensitiveData")

分类打好了，后面就可以按分类来搜索数据资产，如果业务上有脱敏要求，这些标签就是筛选依据。

4.3 建立表与表之间的血缘关系

Atlas的血缘功能可以直观展示数据从哪来到哪去。比如dws_order_summary是由ods_order_detail汇总而来的，那就在这两张表之间建一条血缘关系。

血缘本质上是创建一条“数据转换过程”的实体，类型是Process。我们用一个简单的脚本演示。

# 技术栈：Python 3 + requests

def create_lineage(input_guid, output_guid, process_qualified_name):
    """
    创建一条简单的血缘关系：input -> process -> output
    :param input_guid: 上游表的GUID
    :param output_guid: 下游表的GUID
    :param process_qualified_name: 处理过程的唯一名称
    """
    process_entity = {
        "entity": {
            "typeName": "Process",
            "attributes": {
                "qualifiedName": process_qualified_name,
                "name": "订单日汇总",
                "description": "每天晚上把明细按日期汇总成结果表",
                "inputs": [{"guid": input_guid, "typeName": "hive_table"}],
                "outputs": [{"guid": output_guid, "typeName": "hive_table"}]
            }
        }
    }
    atlas_post("/entity", process_entity)
    print(f"血缘关系 {process_qualified_name} 创建成功")

# 调用示例（参数是假GUID，后续可替换成真实值）
# create_lineage("guid-ods-detail", "guid-dws-summary", "process_ods_to_dws_daily")

血缘关系建好后，你去Atlas UI里打开dws_order_summary，就能看到它的上游是ods_order_detail，链路很清晰。

五、用Atlas API做盘点查询

资产盘点主要靠查询。我们可以用Atlas的搜索接口，把符合条件的表全部列出来。

5.1 查找所有Hive表

Atlas里提供了一套DSL搜索语法，同时也支持简单的type过滤。下面代码演示如何获取所有hive_table类型的数据。

# 技术栈：Python 3 + requests

def get_all_hive_tables():
    """
    获取所有Hive表的基本信息
    :return: 表列表
    """
    # 先查出所有hive_table实体的基本信息
    search_body = {
        "query": "from hive_table select name, owner, qualifiedName"
    }
    # Atlas搜索接口是POST /search/dsl
    url = "/search/dsl"
    response = requests.post(
        f"{ATLAS_BASE_URL}/{url}",
        auth=AUTH,
        headers=HEADERS,
        data=json.dumps(search_body)
    )
    response.raise_for_status()
    data = response.json()
    results = []
    # 搜索结果在 entities 字段里
    for entity in data.get("entities", []):
        attrs = entity.get("attributes", {})
        results.append({
            "name": attrs.get("name"),
            "qualifiedName": attrs.get("qualifiedName"),
            "owner": attrs.get("owner")
        })
        print(f"发现表：{attrs.get('qualifiedName')}，负责人：{attrs.get('owner')}")
    return results

get_all_hive_tables()

这段代码执行后，就能看到一个表清单。如果你的Atlas里已经同步了生产上的Hive元数据，那这个清单就是最原始的资产底册。

5.2 按分类搜索敏感资产

我们刚才给表打了“敏感数据”分类。现在通过分类筛选，快速找出所有敏感表。

Atlas的分类搜索接口可以通过entityClassification属性过滤。使用DSL也行，但更推荐使用简单的查询参数。

# 技术栈：Python 3 + requests

def search_by_classification(classification_name):
    """
    根据分类名搜索所有带该分类的实体
    :param classification_name: 分类名，比如 SensitiveData
    :return: 实体列表
    """
    # 使用基本搜索接口，带classification参数
    params = {
        "typeName": "hive_table",   # 只搜表类型
        "classification": classification_name  # 指定分类
    }
    data = atlas_get("/search/basic", params=params)
    for entity in data.get("entities", []):
        attrs = entity.get("attributes", {})
        print(f"敏感数据表：{attrs.get('qualifiedName')}")

# 调用示例
# search_by_classification("SensitiveData")

这样盘点敏感数据就快多了。甚至还能结合分页接口，把结果导到Excel里做归档。

5.3 查看一张表的前世今生

血缘查询是Atlas的看家本领。通过接口可以拿到某个表的上游和下游。

# 技术栈：Python 3 + requests

def get_table_lineage(guid):
    """
    查询指定实体的血缘信息
    :param guid: 表实体的GUID
    :return: 血缘关系字典
    """
    # 调用Atlas血缘接口
    data = atlas_get(f"/lineage/{guid}")
    # 打印上下流信息
    print("=== 上游表 ===")
    for node in data.get("guidEntityMap", {}).values():
        # 这里简化处理，实际上要根据edges判断方向
        # 为了直观，我们简单输出所有节点
        print(f"节点类型：{node.get('typeName')}，名称：{node.get('attributes', {}).get('qualifiedName')}")
    # 获取edges，展示连接关系
    for edge in data.get("edges", []):
        print(f"边：{edge.get('from')} -> {edge.get('to')}")
    return data

# 调用：get_table_lineage("真实GUID")

血缘数据拿到之后，就可以画一张链路图。线下排查数据质量问题时，照着这个图一步步往上找，很快就能定位到出错环节。

六、关联技术：如何自动同步元数据

手工调用API创建实体毕竟只适合小规模演示。真正落到生产环境，必须让元数据自动同步。Atlas最常用的同步方式是和Hive Metastore做集成。简单说，就是当Hive里执行了CREATE TABLE或ALTER TABLE命令后，Atlas能自动感知并更新元数据。

做这个集成需要改动Hive的配置，把Atlas的Hook加进去。以Hive为例，核心配置如下。

{
  "hive.exec.pre.hooks": "org.apache.atlas.hive.hook.HiveHook",
  "hive.exec.post.hooks": "org.apache.atlas.hive.hook.HiveHook",
  "hive.exec.failure.hooks": "org.apache.atlas.hive.hook.HiveHook"
}

配置了Hook之后，Hive里的建表操作会实时发送到Atlas，血缘也能在跑SQL时自动记录。这样咱们再也不用担心手动创建的元数据不新鲜了。

除了Hive，Atlas还支持Kafka、Sqoop等很多组件。如果你的数据链路里有Kafka消息，也可以让Atlas把Kafka主题当成一种元数据对象来管理，这样消息队列里的数据资产也能纳入盘点范围。

七、应用场景与优缺点

7.1 典型应用场景

数据资产目录建设是最大用途。很多公司会基于Atlas搭建一个内部数据门户，让业务同学能自己查有哪些表可用，而不是整天问数据组要。

数据合规也是一大场景。通过标注敏感分类，可以快速圈出所有涉及个人信息的数据表，方便做安全审计和脱敏改造。比如当法律要求下架某个用户的全部数据时，先通过分类找到相关表，再顺着血缘找到所有衍生表。

数据质量追踪也离不开血缘。某次报表数值不对劲，从报表往下游找，或者从上游找，能大大缩小排查范围。

7.2 优点

开源免费，社区也比较活跃，很多大厂都在用。
有UI界面可以直接浏览，不懂代码的人也能上手。
类型系统相当灵活，可以自定义任何数据类型的元数据。
血缘能力强，支持多种组件的自动采集。

7.3 缺点

部署和运维有门槛，因为它依赖Hadoop生态组件，本身对机器配置要求不低。
UI交互做得一般，在数据量大时页面会有卡顿感。
查询性能需要优化，因为元数据都放在JanusGraph里，数据量大了之后，搜索可能变慢。
API文档不太友好，学习成本稍高，这也是咱们写这篇博客的一个原因。

八、注意事项与建议

做数据资产盘点不能指望一上来就全自动。初期建议先梳理最核心的库表，比如业务库的宽表、报表依赖的中间表，把这些先录入Atlas，形成一份最小可用目录。然后再逐渐扩大范围。

字段命名规范性很重要。如果一张表的字段名叫c1、c2，就算录进Atlas也没人看懂。所以在盘点之前，最好先推动团队把表的注释补全，这些注释会同步成Atlas里的描述信息。

定期做元数据健康检查。可以写个定时任务，每天检查Atlas里是否有实体缺失了所有者信息、表描述为空。对于这些不完整的记录，标记为“待完善”，并反馈给负责人。

权限控制不要忘。Atlas自带权限管理，但很多使用者会忽略。按业务线划分好用户组，给不同的人分配不同的查看和编辑权限。毕竟资产目录也是数据的一部分，不能谁都能乱改。

九、总结

Apache Atlas是个很实用的元数据管理工具，能帮我们把散落的数据资产变成一份可搜索、可追踪的清单。今天从创建表实体、添加字段、打分类标签、建血缘关系，到搜索和查询，完整走了一遍。你会发现核心思路很简单：让Atlas知道你有哪些数据，这些数据从哪里来，数据是否敏感。一切都是通过API完成，所以也方便自动化。

当然，工具只是辅助，真正的盘点需要业务方、数据开发、运维一起参与。先把一份小范围的资产目录建起来，再用Atlas持续补充和维护，数据团队就不再怕“我们有哪些表”这种灵魂拷问了。

围绕“利用Apache Atlas进行数据资产盘点的实操方法与工具使用”参与讨论

0 条

利用Apache Atlas进行数据资产盘点的实操方法与工具使用

一、为什么需要做数据资产盘点

二、Atlas能帮我们做什么

三、实操前的准备工作

3.1 跑起一个Atlas实例

3.2 技术栈说明

3.3 先给自己发一张“门禁卡”

四、把数据源信息喂给Atlas

4.1 创建表类型

4.2 给数据打上分类标签

4.3 建立表与表之间的血缘关系

五、用Atlas API做盘点查询

5.1 查找所有Hive表

5.2 按分类搜索敏感资产

5.3 查看一张表的前世今生

六、关联技术：如何自动同步元数据

七、应用场景与优缺点

7.1 典型应用场景

7.2 优点

7.3 缺点

八、注意事项与建议

九、总结

评论

关联文章

AI助手

一、为什么需要做数据资产盘点

二、Atlas能帮我们做什么

三、实操前的准备工作

3.1 跑起一个Atlas实例

3.2 技术栈说明

3.3 先给自己发一张“门禁卡”

四、把数据源信息喂给Atlas

4.1 创建表类型

4.2 给数据打上分类标签

4.3 建立表与表之间的血缘关系

五、用Atlas API做盘点查询

5.1 查找所有Hive表

5.2 按分类搜索敏感资产

5.3 查看一张表的前世今生

六、关联技术：如何自动同步元数据

七、应用场景与优缺点

7.1 典型应用场景

7.2 优点

7.3 缺点

八、注意事项与建议

九、总结

同分类面试题

关联文章

AI助手