Apache Atlas 入门教程：从零开始掌握企业级数据治理平台

┌─────────────────────────────────────────────────────────────┐
│                        客户端层                              │
│         Web UI    REST API    Java API    Python API         │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                     Atlas 核心服务层                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │  元数据仓库  │  │  图数据库   │  │    搜索与索引引擎    │  │
│  │  (Type System)│ │  (JanusGraph)│ │    (Elasticsearch)  │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                      集成层                                  │
│   Hive Hook   Spark Hook   Kafka Hook   HBase Hook          │
│   Sqoop Hook  Storm Hook   Falcon Hook  Custom Hook         │
└─────────────────────────────────────────────────────────────┘

# 下载 Atlas 二进制包
wget https://downloads.apache.org/atlas/2.3.0/apache-atlas-2.3.0-server.tar.gz

# 解压
tar -zxvf apache-atlas-2.3.0-server.tar.gz

# 进入安装目录
cd apache-atlas-2.3.0

apache-atlas-2.3.0/
├── bin/          # 启动脚本
├── conf/         # 配置文件
├── hbase/        # 内嵌 HBase（仅测试环境使用）
├── hook/         # 各组件 Hook
├── hook-bin/     # Hook 管理工具
├── lib/          # 依赖库
├── logs/         # 日志目录
├── model/        # 内置类型定义
└── server/       # Web 应用

#########################################
# Atlas 服务器基本配置
#########################################
atlas.rest.address=http://localhost:21000

# 启用认证
atlas.authentication.method.file=true
atlas.authentication.method.file.filename=${sys:atlas.home}/conf/users-credentials.properties

#########################################
# 图数据库配置 (使用嵌入式 HBase)
#########################################
atlas.graph.storage.backend=hbase2
atlas.graph.storage.hbase.table=apache_atlas_janus
atlas.graph.storage.hostname=localhost

#########################################
# 索引引擎配置 (使用嵌入式 Solr)
#########################################
atlas.graph.index.search.backend=solr
atlas.graph.index.search.solr.mode=http
atlas.graph.index.search.solr.http-urls=http://localhost:2181

#########################################
# Kafka 配置
#########################################
atlas.notification.embedded=false
atlas.kafka.bootstrap.servers=localhost:9092
atlas.kafka.zookeeper.connect=localhost:2181
atlas.kafka.zookeeper.session.timeout.ms=400
atlas.kafka.zookeeper.connection.timeout.ms=200
atlas.kafka.enable.auto.commit=true

#########################################
# 安全配置
#########################################
atlas.ssl.exclude.cipher.suites=.*_GCM_.*,TLS_RSA.*

# 设置环境变量
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export ATLAS_HOME=/opt/apache-atlas-2.3.0

# 启动 Atlas（自动启动内嵌的 HBase 和 Solr）
${ATLAS_HOME}/bin/atlas_start.py

# 检查启动状态
${ATLAS_HOME}/bin/atlas_admin.py -status

# JVM 内存配置（在 atlas-env.sh 中设置）
export ATLAS_SERVER_HEAP="-Xms4g -Xmx8g -XX:MaxNewSize=1g"

# 增加图数据库连接池大小
atlas.graph.storage.connection.timeout-ms=10000

DataSet（抽象基类）
├── hive_db
├── hive_table
│   └── hive_column
├── hbase_namespace
│   └── hbase_table
│       └── hbase_column_family
├── kafka_topic
└── rdbms_db
    └── rdbms_table
        └── rdbms_column

Process（抽象基类）
├── hive_process
├── hive_column_lineage
└── spark_process

{
  "entityDefs": [
    {
      "name": "custom_dataset",
      "superTypes": ["DataSet"],
      "typeVersion": "1.0",
      "description": "自定义数据集类型",
      "attributeDefs": [
        {
          "name": "owner",
          "typeName": "string",
          "isOptional": false,
          "cardinality": "SINGLE",
          "description": "数据集负责人"
        },
        {
          "name": "dataClassification",
          "typeName": "string",
          "isOptional": true,
          "cardinality": "SINGLE",
          "description": "数据分级（公开/内部/机密/绝密）"
        },
        {
          "name": "updateFrequency",
          "typeName": "string",
          "isOptional": true,
          "cardinality": "SINGLE",
          "description": "数据更新频率"
        }
      ]
    }
  ]
}

{
  "classificationDefs": [
    {
      "name": "PII",
      "description": "个人身份信息",
      "superTypes": [],
      "attributeDefs": [
        {
          "name": "piiType",
          "typeName": "string",
          "isOptional": false,
          "description": "PII 类型（姓名/手机号/身份证/邮箱等）"
        }
      ]
    },
    {
      "name": "FINANCIAL_DATA",
      "description": "财务敏感数据",
      "superTypes": [],
      "attributeDefs": []
    }
  ]
}

curl -X POST \
  http://localhost:21000/api/atlas/v2/entity/guid/{entity_guid}/classifications \
  -H 'Content-Type: application/json' \
  -u admin:admin \
  -d '[
    {
      "typeName": "PII",
      "attributes": {
        "piiType": "手机号"
      },
      "propagate": true,
      "removePropagationsOnEntityDelete": false
    }
  ]'

hive_table: raw_orders  ──► hive_process: etl_transform  ──► hive_table: cleaned_orders
                                                                      │
                                                                      ▼
                                                         hive_process: agg_by_day
                                                                      │
                                                                      ▼
                                                         hive_table: daily_sales_report

# 查询实体的血缘图（默认深度为3）
curl -X GET \
  "http://localhost:21000/api/atlas/v2/lineage/{guid}?direction=BOTH&depth=5" \
  -u admin:admin

# 创建实体
curl -X POST \
  http://localhost:21000/api/atlas/v2/entity \
  -H 'Content-Type: application/json' \
  -u admin:admin \
  -d '{
    "entity": {
      "typeName": "hive_table",
      "attributes": {
        "name": "user_info",
        "qualifiedName": "default.user_info@cluster1",
        "db": {
          "typeName": "hive_db",
          "uniqueAttributes": {
            "qualifiedName": "default@cluster1"
          }
        },
        "description": "用户基本信息表",
        "owner": "data_team",
        "tableType": "MANAGED"
      }
    }
  }'

# 根据 GUID 查询实体
curl -X GET \
  http://localhost:21000/api/atlas/v2/entity/guid/{guid} \
  -u admin:admin

# 根据唯一属性查询实体
curl -X GET \
  "http://localhost:21000/api/atlas/v2/entity/uniqueAttribute/type/hive_table?attr:qualifiedName=default.user_info@cluster1" \
  -u admin:admin

# 更新实体属性
curl -X PUT \
  http://localhost:21000/api/atlas/v2/entity/guid/{guid}/attribute?name=description \
  -H 'Content-Type: application/json' \
  -u admin:admin \
  -d '"更新后的描述信息"'

# 删除实体（软删除）
curl -X DELETE \
  http://localhost:21000/api/atlas/v2/entity/guid/{guid} \
  -u admin:admin

# 基本搜索 - 按类型和属性搜索
curl -X POST \
  http://localhost:21000/api/atlas/v2/search/basic \
  -H 'Content-Type: application/json' \
  -u admin:admin \
  -d '{
    "typeName": "hive_table",
    "classification": "PII",
    "limit": 25,
    "offset": 0,
    "sortBy": "name",
    "sortOrder": "ASCENDING"
  }'

# 全文搜索
curl -X GET \
  "http://localhost:21000/api/atlas/v2/search/fulltext?query=user_info&limit=10" \
  -u admin:admin

# DSL 搜索（使用 Atlas 查询语言）
curl -X GET \
  "http://localhost:21000/api/atlas/v2/search/dsl?query=hive_table+where+name+%3D+%27user_info%27" \
  -u admin:admin

pip install apache-atlas

from apache_atlas.client.atlas import AtlasClient
from apache_atlas.model.entity import AtlasEntity, AtlasEntitiesWithExtInfo

# 初始化客户端
client = AtlasClient('http://localhost:21000', ('admin', 'admin'))

# 查询实体
entity = client.entity.get_entity_by_attribute(
    type_name='hive_table',
    uniq_attributes={'qualifiedName': 'default.user_info@cluster1'}
)
print(f"Entity GUID: {entity.entity.guid}")
print(f"Entity Name: {entity.entity.attributes.get('name')}")

# 创建新实体
new_entity = AtlasEntity()
new_entity.typeName = 'hive_table'
new_entity.attributes = {
    'name': 'order_detail',
    'qualifiedName': 'sales_db.order_detail@prod_cluster',
    'description': '订单明细表',
    'owner': 'order_team'
}

entities_info = AtlasEntitiesWithExtInfo()
entities_info.entities = [new_entity]

response = client.entity.create_entities(entities_info)
print(f"Created entity GUID: {response.guidAssignments}")

# 搜索实体
from apache_atlas.model.search import SearchParameters
search_params = SearchParameters()
search_params.typeName = 'hive_table'
search_params.limit = 10
search_params.offset = 0

search_result = client.discovery.faceted_search(search_params)
for entity in search_result.entities or []:
    print(f"  - {entity.displayText} ({entity.guid})")

# 将 Atlas Hook 相关 JAR 复制到 Hive 的 lib 目录
cp ${ATLAS_HOME}/hook/hive/*.jar ${HIVE_HOME}/lib/

<configuration>
  <!-- 启用 Atlas Hook -->
  <property>
    <name>hive.exec.post.hooks</name>
    <value>org.apache.atlas.hive.hook.HiveHook</value>
  </property>
  
  <!-- Atlas Server 地址 -->
  <property>
    <name>atlas.hook.hive.synchronous</name>
    <value>false</value>
  </property>
</configuration>

atlas.zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
atlas.kafka.bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
atlas.hook.hive.synchronous=false
atlas.hook.hive.numRetries=3
atlas.hook.hive.queueSize=10000

-- 创建数据库
CREATE DATABASE sales_db COMMENT '销售业务数据库';

-- 创建数据表
CREATE TABLE sales_db.orders (
  order_id    BIGINT    COMMENT '订单ID',
  user_id     BIGINT    COMMENT '用户ID',
  order_date  STRING    COMMENT '下单日期',
  total_amount DECIMAL(10,2) COMMENT '订单总金额',
  status      STRING    COMMENT '订单状态'
) COMMENT '订单主表'
STORED AS ORC;

-- 执行数据转换操作（会产生血缘关系）
CREATE TABLE sales_db.daily_sales AS
SELECT 
  order_date,
  COUNT(*) as order_count,
  SUM(total_amount) as daily_total
FROM sales_db.orders
WHERE status = 'COMPLETED'
GROUP BY order_date;

# 导入所有数据库和表
${ATLAS_HOME}/hook-bin/import-hive.sh

# 导入指定数据库
${ATLAS_HOME}/hook-bin/import-hive.sh -d sales_db

# 导入指定表
${ATLAS_HOME}/hook-bin/import-hive.sh -d sales_db -t orders

<系统名>.<对象名>@<集群名>

from apache_atlas.client.atlas import AtlasClient
from apache_atlas.model.entity import AtlasEntity, AtlasEntitiesWithExtInfo

client = AtlasClient('http://localhost:21000', ('admin', 'admin'))

# 创建 Process 实体来记录血缘
process_entity = AtlasEntity()
process_entity.typeName = 'Process'
process_entity.attributes = {
    'name': 'custom_etl_job_001',
    'qualifiedName': 'custom_etl.job_001@prod',
    'description': '自研 ETL 作业：将原始日志转换为用户行为分析表',
    'inputs': [
        {
            'typeName': 'hive_table',
            'uniqueAttributes': {
                'qualifiedName': 'raw_db.user_logs@prod'
            }
        }
    ],
    'outputs': [
        {
            'typeName': 'hive_table',
            'uniqueAttributes': {
                'qualifiedName': 'analytics_db.user_behavior@prod'
            }
        }
    ]
}

entities_info = AtlasEntitiesWithExtInfo()
entities_info.entities = [process_entity]
client.entity.create_entities(entities_info)
print("血缘关系注册成功！")

# 调整 Elasticsearch 索引配置
atlas.graph.index.search.elasticsearch.index.number_of_shards=5
atlas.graph.index.search.elasticsearch.index.number_of_replicas=1

# 增加搜索结果缓存
atlas.search.maxlimit=1000
atlas.search.defaultlimit=100

# 批量提交配置
atlas.graph.storage.buffer-size=2048
atlas.graph.storage.cache.db-cache=true
atlas.graph.storage.cache.db-cache-clean-wait=20
atlas.graph.storage.cache.db-cache-size=0.5

atlas.hook.hive.queueSize=100000
atlas.hook.hive.numRetries=5
atlas.notification.hook.asynchronous=true

# 检查 Atlas 服务状态
curl -X GET http://localhost:21000/api/atlas/admin/status -u admin:admin

# 查看版本信息
curl -X GET http://localhost:21000/api/atlas/admin/version -u admin:admin

# 查看统计信息
curl -X GET http://localhost:21000/api/atlas/admin/metrics -u admin:admin

# 导出 Atlas 元数据快照
${ATLAS_HOME}/bin/atlas_admin.py -exportData output_dir/

# 恢复元数据
${ATLAS_HOME}/bin/atlas_admin.py -importData input_dir/

# 检查 Kafka 消息
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group atlas

# 重建搜索索引
curl -X PUT \
  http://localhost:21000/api/atlas/admin/metrics/enableMetrics \
  -u admin:admin