【Atlas】治理工作流（如标签审批）是否原生支持？

Apache Atlas 2.4.0 治理工作流（标签审批）能力边界与金融级扩展实战

用户问题原文：治理工作流（如标签审批）是否原生支持？

本文将直面这一高频误解，系统性地澄清Apache Atlas 2.4.0 在治理工作流领域的实际能力边界，并基于金融交易流水表finance_tx_lineage的 PII 标签审批需求，构建一套以 Kafka + Airflow 为核心的生产级审批扩展方案。我们将通过源码剖析、架构图、配置示例与验证命令，揭示“元数据平台如何与工作流引擎协同实现标签审批”的正确打开方式。

一、问题引入：谁允许分析师直接标记银行卡号为非敏感？

某大型银行的数据治理团队发现严重合规漏洞：普通数据分析师账号analyst_user直接通过 Atlas UI 将finance_tx_lineage.card_number字段的PCI.CARD_NUMBER标签删除，导致该字段在后续查询中以明文返回。

团队需要立即实施：

  • 标签变更必须经过数据治理专员审批；
  • 高风险标签（如 PCI/PII）禁止自助操作。

根本原因在于：Apache Atlas 2.4.0 本身不提供任何内置的审批工作流引擎。所有 Classification 操作均为即时生效。

关键界定：

  • “治理工作流”指Classification 变更需经过多步骤审批（如提交→审核→生效）的流程控制机制。
    结论前置：Atlas 原生不支持审批工作流，必须通过外部系统拦截 REST API 并实现状态机管理。

二、原理解析：Atlas 的即时生效模型与设计哲学

2.1 官方立场与源码佐证

Apache Atlas 项目从未将工作流引擎纳入核心功能。其设计哲学是“元数据存储与查询平台”而非“治理流程引擎”。

  • GitHub Issue 明确表态：
    ATLAS-3215 中，社区讨论审批功能，Committer 回复：

    “Workflow capabilities are out of scope for Atlas core. Consider integrating with external workflow systems.”

  • 源码结构验证：
    在apache/atlas仓库中，Classification 相关 API 均为直接写入 JanusGraph（见repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphMapper.java），无任何状态机或审批逻辑。

通俗类比：
Atlas 的标签操作就像超市自助结账机——你扫描商品（选择标签），机器立即扣款（写入元数据），没有收银员检查（审批）环节。
技术本质差异：自助结账依赖用户诚信；而金融治理要求强制审批流程。

2.2 即时生效的操作类型

Atlas 2.4.0 所有 Classification 操作均为原子且即时：

操作REST API是否可拦截
添加标签POST /entity/guid/{guid}/classification否（除非代理层拦截）
删除标签DELETE /entity/guid/{guid}/classification/{name}否
更新标签属性PUT /entity/guid/{guid}/classification否

关键限制：
一旦请求到达 Atlas Server，变更立即生效。无法在 Atlas 内部实现“待审批”状态。

三、架构全景：金融级标签审批流水线

3.1 整体架构图（Mermaid）

Submit Tag Request

Validate & Enqueue

Consume Requests

Create Approval Task

Approve/Reject

Send Decision

Consume Decisions

Call Atlas REST API

User: analyst_user

API Gateway

Kafka Topic: tag_approval_requests

Airflow DAG

Slack/Email

Governor: data_governor

Kafka Topic: tag_approval_decisions

Approval Processor

Atlas Server

颜色说明：

  • #333：用户
  • #f96：审批工作流
  • #00f：Atlas 核心
  • #0f0：通知系统

3.2 核心组件职责

组件职责技术选型
API Gateway拦截原始 Atlas API，转为审批请求Kong / Nginx
Kafka存储审批请求与决策Apache Kafka 3.3+
Airflow编排审批任务与超时处理Apache Airflow 2.7+
Slack/Email通知审批人Webhook / SMTP
Approval Processor执行最终 Atlas 操作自定义 Java 服务

四、实战配置：构建金融交易流水的标签审批链路

4.1 步骤 1：部署 API Gateway 拦截层

4.1.1 Kong 配置示例
# 创建 Atlas Upstreamcurl-XPOST http://kong:8001/upstreams\--dataname=atlas-upstream\--datahost=atlas-server# 创建 Servicecurl-XPOST http://kong:8001/services\--dataname=atlas-service\--dataurl='http://atlas-upstream:21000'# 创建 Route（拦截 Classification API）curl-XPOST http://kong:8001/services/atlas-service/routes\--data'paths[]=/api/atlas/v2/entity/guid/.*/classification'\--dataname=tag-approval-route
4.1.2 自定义插件（Lua）
-- tag-approval.lualocalfunctionhandle_classification_request(conf,ctx)localpath=ctx.req.get_path()localmethod=ctx.req.get_method()-- 仅拦截非治理专员的操作localuser=ctx.req.get_header("X-Forwarded-User")ifnotstring.match(user,"data_governor")then-- 转发到审批队列send_to_kafka({user=user,operation=method,path=path,timestamp=ngx.time()})-- 返回 202 Acceptedctx.resp.set_status(202)ctx.resp.set_body('{"status": "pending_approval"}')returnkong.response.exit(202)end-- 治理专员直通returnkong.service.request.set_header("Authorization","Basic YWRtaW46YWRtaW4=")end

⚠️危险操作警告：
必须严格验证用户身份（通过 Kerberos/LDAP），防止绕过审批。

4.2 步骤 2：实现 Airflow 审批 DAG

# dags/tag_approval.pyfromairflowimportDAGfromairflow.operators.pythonimportPythonOperatorfromdatetimeimporttimedeltaimportrequestsdefcreate_approval_task(**context):request=context['dag_run'].conf user=request['user']entity_guid=extract_guid(request['path'])# 发送 Slack 通知slack_msg=f"用户{user}请求修改实体{entity_guid}的标签，请审批！"requests.post(SLACK_WEBHOOK,json={"text":slack_msg})# 设置审批超时（24小时）context['task_instance'].set_duration(timedelta(hours=24))defexecute_approval(**context):decision=context['dag_run'].conf['decision']ifdecision=='approved':# 调用 Atlas API 执行原始操作atlas_api_call(context['dag_run'].conf['original_request'])withDAG('tag_approval',schedule_interval=None)asdag:create_task=PythonOperator(task_id='create_approval',python_callable=create_approval_task)execute_task=PythonOperator(task_id='execute_approval',python_callable=execute_approval)create_task>>execute_task

4.3 步骤 3：模拟审批全流程

4.3.1 用户提交删除请求
# 通过 API Gateway 提交（非治理专员）curl-uanalyst_user:password-XDELETE\http://kong:8000/api/atlas/v2/entity/guid/<card_guid>/classification/PCI.CARD_NUMBER# 返回：{"status": "pending_approval"}
4.3.2 治理专员审批
// 在 Slack 点击“批准”按钮，触发 Webhook{"decision":"approved","original_request":{"method":"DELETE","path":"/api/atlas/v2/entity/guid/<card_guid>/classification/PCI.CARD_NUMBER"}}
4.3.3 验证标签变更
# 检查标签是否已删除curl-uadmin:admin\"http://atlas:21000/api/atlas/v2/entity/guid/<card_guid>/classifications"# 预期：返回空数组 []

五、高级特性：分级审批与自动拒绝

5.1 高风险标签自动拒绝

  • 规则：PCI/PII 标签不允许删除，仅允许添加。
  • 实现：在 API Gateway 插件中增加规则引擎：
    ifoperation=="DELETE"andstring.match(classification,"PCI.*")thenctx.resp.set_status(403)ctx.resp.set_body('{"error": "PCI tags cannot be deleted"}')returnkong.response.exit(403)end

5.2 多级审批

  • 场景：删除 GDPR 相关标签需法务 + 治理双审批。
  • 实现：Airflow DAG 中增加多个审批任务节点。

六、FAQ：高频关联问题解答

Q1：能否在 Atlas UI 中集成审批？

能，但需定制 UI：

  • 替换默认的 Classification 操作按钮为“提交审批”；
  • 新增“我的审批”页面。

Q2：审批期间如何显示状态？

  • 方案：在 Entity 上添加临时 ClassificationPENDING_APPROVAL；
  • 注意：需确保该标签不影响 Ranger 策略。

Q3：性能影响如何？

  • 延迟增加：审批流程增加分钟级延迟；
  • 优化：低风险操作（如添加非敏感标签）可豁免审批。

Q4：云上如何实现？

  • AWS：Step Functions + SNS（替代 Airflow + Slack）；
  • Azure：Logic Apps + Teams。

Q5：如何审计审批记录？

  • 存储：将审批请求与决策存入独立审计表；
  • 查询：通过 Elasticsearch 提供审批历史搜索。

七、总结与最佳实践

  • 金融级合规必备：SOX、PCI DSS 等法规要求关键操作审批留痕，必须通过外部工作流实现。
  • 三层架构是黄金标准：拦截层（API Gateway） + 编排层（Airflow） + 执行层（Processor）。
  • 金融场景最佳实践：
    1. 分级策略：高风险标签强制审批，低风险标签自助操作；
    2. 超时机制：审批超过 24 小时自动拒绝；
    3. 双人审批：涉及 GDPR/CCPA 的操作需两人批准。

避坑指南：

  • 避免在审批期间锁定 Entity（影响其他操作）；
  • 定期清理 Pending 状态的僵尸请求。

作者署名：九师兄

  • 专题目录：【Apache Atlas】Apache Atlas 资深工程师到专家实战之路目录
  • 总目录：【目录】技术体系目录

注意：本文由 AI 辅助生成，技术细节请以官方文档为准。生产环境使用前务必充分测试。