【Atlas】Atlas 的 Notification 消息格式是怎样的（ENTITY_CREATE 等）？

Apache Atlas 2.4.0 Notification 消息格式全解ENTITY_CREATE 等事件的深度剖析与生产应用用户问题原文“92. Atlas 的 Notification 消息格式是怎样的ENTITY_CREATE 等”本文将以前所未有的深度全面解析Apache Atlas 2.4.0中Notification 消息的完整格式。我们将聚焦于ATLAS_HOOK和ATLAS_ENTITIES两个核心 Kafka Topic 中流通的消息结构详细拆解ENTITY_CREATE、ENTITY_UPDATE、CLASSIFICATION_ADD等关键事件类型的 JSON Schema。文章将以IoT 设备指标 Hudi 表 (iot_device_metrics_hudi)的元数据上报与治理为真实场景结合源码、配置和验证命令为你提供一份可直接用于开发消费者或排查线上问题的权威指南。1. 问题引入消息格式不明导致的血缘断裂事故在某 IoT 平台一个自研的Hudi Hook负责将新创建的 Hudi 表元数据上报至 Atlas。然而上线后发现部分表的血缘关系未能正确建立。经过数小时排查根本原因在于Hook 发送到ATLAS_HOOKTopic 的消息中relationshipAttributeDefs字段的格式与 Atlas Server 期望的格式不一致。Server 在反序列化时静默失败导致关系未被创建但主表 Entity 却成功入库造成了“有表无血缘”的诡异现象。这次 P1 级事故凸显了精确理解 Atlas Notification 消息格式的重要性。无论是开发自定义 Hook还是消费ATLAS_ENTITIES事件都必须对消息的每一个字段了如指掌。2. 原理解析Notification 的设计哲学与两大通道2.1 核心概念为什么需要标准化的消息格式官方/源码解释Atlas 的 Notification 机制旨在提供一种松耦合、异步、可靠的元数据变更通信方式。标准化的 JSON 消息格式是实现这一目标的基础它确保了不同版本、不同语言的生产者和消费者之间能够无缝协作。通俗类比Notification 消息就像国际邮政的“标准信封”。无论信件从哪个国家Hive/Flink/自定义系统寄出只要遵循统一的地址格式JSON Schema邮局Kafka就能准确投递收件人Atlas Server/消费者也能正确解读内容。技术本质差异邮政信封的内容是自由的而 Notification 消息的内容结构是严格约束的由 Atlas 的 Java 模型类如EntityNotificationV2强类型定义。2.2 两大消息通道及其角色Atlas 定义了两种主要的通知类型对应两个 Kafka Topic通道Topic 名称生产者消费者消息性质输入通道ATLAS_HOOK外部系统 (Hive, Flink, 自定义 Hook)Atlas Server指令性请求 Atlas 执行某个操作如创建实体输出通道ATLAS_ENTITIESAtlas Server外部系统 (治理服务, 数据地图)宣告性通知外部系统某个操作已成功完成本文将分别解析这两种通道的消息格式。2.3 消息序列化与反序列化的底层机制所有 Notification 消息的序列化和反序列化都通过Jackson库完成。关键的模型类位于intg/src/main/java/org/apache/atlas/model/notifications包下。生产者侧调用NotificationInterface.sendEntityNotification(...)方法内部会使用ObjectMapper将 Java 对象如AtlasEntity转换为 JSON 字符串。消费者侧从 Kafka 读取到 JSON 字符串后使用ObjectMapper.readValue(json, HookNotificationV2.class)将其还原为 Java 对象。源码路径org.apache.atlas.notification.NotificationInterfaceorg.apache.atlas.model.notifications.HookNotificationV2org.apache.atlas.model.instance.AtlasEntity3.ATLAS_HOOK消息格式详解输入通道这是外部系统如 Hive Hook发送给 Atlas Server 的请求消息。3.1 通用消息结构{version:{version:2.4.0},message:{type:ENTITY_CREATE,// 操作类型user:hive,// 触发操作的用户timestamp:1714000000000,// 时间戳 (毫秒)entity:{...}// 具体的实体对象}}3.2ENTITY_CREATE/ENTITY_PARTIAL_UPDATE详细格式当创建或部分更新一个实体时message.entity字段包含一个AtlasEntityWithExtInfo对象。示例创建一张 Hudi 表{version:{version:2.4.0},message:{type:ENTITY_CREATE,user:hudi_user,timestamp:1714000000000,entity:{entity:{typeName:hudi_table,attributes:{qualifiedName:dwd.iot_device_metrics_hudihudi-prod-cluster,name:iot_device_metrics_hudi,description:IoT设备核心指标宽表,owner:data_platform_team,columns:[{typeName:hudi_column,attributes:{qualifiedName:dwd.iot_device_metrics_hudihudi-prod-cluster.device_id,name:device_id}}]},classifications:[// 可选可在创建时附加分类{typeName:PII.NonSensitive,attributes:{}}]},referredEntities:{}// 可选用于内联引用其他实体}}}关键字段说明typeName: 必须是 Atlas Type System 中已注册的类型名。qualifiedName:全局唯一标识是血缘追踪的基石。格式必须符合预定义规则。columns: 是一个数组每个元素都是一个完整的hudi_column实体定义。这实现了嵌套实体的创建。classifications: 允许在创建实体的同时为其打上标签。3.3ENTITY_FULL_UPDATE与ENTITY_DELETEENTITY_FULL_UPDATE: 结构与ENTITY_CREATE相同但要求提供实体的完整快照。Atlas 会用此快照完全覆盖现有实体。ENTITY_DELETE: 消息体极为简单只需提供实体的guid或uniqueAttributes。{message:{type:ENTITY_DELETE,user:admin,entity:{guid:a1b2c3d4-5678-90ab-cdef-1234567890ab}}}4.ATLAS_ENTITIES消息格式详解输出通道这是 Atlas Server 在成功处理变更后广播给所有消费者的宣告性消息。4.1 通用消息结构与ATLAS_HOOK不同ATLAS_ENTITIES的顶层是一个数组可以一次包含多个实体的变更。{version:{version:2.4.0},msgCompressionKind:NONE,msgSplitIdx:0,msgSplitCount:1,msgSourceIP:10.0.0.1,entities:[// 注意这里是数组{...},// 第一个实体变更{...}// 第二个实体变更]}4.2 单个实体变更 (EntityNotification) 详细格式数组中的每个元素都是一个EntityNotification对象。示例宣告一张 Hudi 表已创建{entities:[{entity:{typeName:hudi_table,attributes:{qualifiedName:dwd.iot_device_metrics_hudihudi-prod-cluster,name:iot_device_metrics_hudi,description:IoT设备核心指标宽表,owner:data_platform_team},guid:e1f2g3h4-5678-90ab-cdef-1234567890ab,// Atlas分配的全局唯一IDstatus:ACTIVE,// 状态: ACTIVE, DELETEDcreatedBy:hudi_user,updatedBy:hudi_user,createTime:1714000000000,updateTime:1714000000000,version:0,// 版本号每次更新递增relationshipAttributes:{// 关系属性在此处展开columns:[{guid:i1j2k3l4-5678-90ab-cdef-1234567890ab,typeName:hudi_column,uniqueAttributes:{qualifiedName:dwd.iot_device_metrics_hudihudi-prod-cluster.device_id}}],db:{guid:...,typeName:hudi_db,uniqueAttributes:{qualifiedName:dwdhudi-prod-cluster}}}},operationType:ENTITY_CREATE,// 关键变更类型eventTime:1714000000123// 事件发生时间}]}关键字段说明operationType: 这是消费者进行逻辑分支的核心依据。可能的值包括ENTITY_CREATEENTITY_UPDATEENTITY_DELETECLASSIFICATION_ADDCLASSIFICATION_UPDATECLASSIFICATION_DELETErelationshipAttributes: 与ATLAS_HOOK中的内联定义不同这里的关系是以引用Reference的形式存在的只包含guid和uniqueAttributes避免了消息体过大。guid: 由 Atlas Server 在创建实体时生成是后续所有操作更新、删除、打标的唯一凭证。4.3 分类变更 (CLASSIFICATION_ADD等) 消息格式当一个分类被添加到实体上时消息格式如下{entities:[{entity:{guid:e1f2g3h4-5678-90ab-cdef-1234567890ab,typeName:hudi_table},classification:{typeName:PII.Sensitive,attributes:{expiryDate:2030-12-31},entityGuid:e1f2g3h4-5678-90ab-cdef-1234567890ab},operationType:CLASSIFICATION_ADD,eventTime:1714000005000}]}注意此时entity字段只包含guid和typeName完整的实体信息需要消费者通过 REST API 查询。5. 实战捕获并解析 Hudi 表创建事件5.1 步骤一通过 REST API 创建测试实体curl-uadmin:admin-XPOST-HContent-Type: application/json\-d{ entity: { typeName: hudi_table, attributes: { qualifiedName: dwd.iot_device_metrics_hudihudi-prod-cluster, name: iot_device_metrics_hudi } } }http://atlas-server:21000/api/atlas/v2/entity5.2 步骤二消费ATLAS_ENTITIES验证格式kafka-console-consumer.sh --bootstrap-server localhost:9092--topicATLAS_ENTITIES --from-beginning|jq.# 验证点输出应包含一个 entities 数组其中 operationType 为 ENTITY_CREATE# entity.guid 不为空且 relationshipAttributes.db 存在。5.3 步骤三Java 代码解析示例使用 Atlas 官方模型类进行安全反序列化。importorg.apache.atlas.model.notifications.HookNotificationV2;importcom.fasterxml.jackson.databind.ObjectMapper;publicclassNotificationParser{privatestaticfinalObjectMapperMAPPERnewObjectMapper();publicstaticvoidmain(String[]args)throwsException{// 假设 messageJson 是从Kafka读取的字符串StringmessageJson...;// 1. 安全反序列化HookNotificationV2notificationMAPPER.readValue(messageJson,HookNotificationV2.class);// 2. 遍历所有变更for(HookNotificationV2.EntityNotificationentityNotif:notification.getEntities()){StringopTypeentityNotif.getOperationType().name();AtlasEntityentityentityNotif.getEntity();if(ENTITY_CREATE.equals(opType)hudi_table.equals(entity.getTypeName())){Stringguidentity.getGuid();Stringqn(String)entity.getAttribute(qualifiedName);System.out.println(New Hudi table created! GUID: guid, QN: qn);// 3. 访问关系属性MapString,ObjectrelAttrsentity.getRelationshipAttributes();if(relAttrs!null){ObjectdbRefrelAttrs.get(db);// dbRef 是一个 AtlasObjectId 对象}}}}}⚠️重要警告切勿使用通用的MapString, Object来解析消息这会导致类型错误和难以维护的代码。始终使用 Atlas 提供的强类型模型类。6. FAQ 与最佳实践FAQQ: 消息中的version字段有何作用A: 主要用于向后兼容。当 Atlas 升级导致消息格式变更时消费者可以根据version字段决定如何解析。目前 2.4.0 的版本号就是 “2.4.0”。Q:msgSplitIdx和msgSplitCount是干什么的A: 当单条消息过大超过 Kafka 的max.message.bytes时Atlas 会自动将其分片Split发送。msgSplitCount表示总片数msgSplitIdx表示当前片的索引从0开始。消费者需要自行拼接。但在绝大多数场景下这两个值分别是 1 和 0。Q:ATLAS_HOOK和ATLAS_ENTITIES的消息格式为何不同A: 因为它们的语义不同。ATLAS_HOOK是一个请求需要包含完整的操作指令和实体定义。ATLAS_ENTITIES是一个事件宣告侧重于告知“什么变了”并且为了效率关系属性采用引用而非内联。Q: 如何查看 Atlas 内部使用的完整 JSON SchemaA: 最权威的方式是查阅atlas-intg模块的 Java 源码特别是AtlasEntity,HookNotificationV2等类。这些类上的 Jackson 注解如JsonProperty定义了最终的 JSON 结构。Q: 能否自定义 Notification 消息的格式A:不能。消息格式是由 Atlas Server 的反序列化逻辑硬编码的。任何不符合格式的消息都会被丢弃并在 Server 日志中记录ERROR。自定义 Hook 必须严格遵守官方格式。监控建议生产者监控监控自定义 Hook 的日志确保没有SerializationException。消费者监控监控消费者的反序列化失败率。任何无法解析的消息都应进入死信队列DLQ以便人工分析。Kafka 监控监控ATLAS_HOOK和ATLAS_ENTITIESTopic 的生产/消费速率、积压情况。生产最佳实践Schema 验证在 CI/CD 流程中加入对 Hook 产出消息的 Schema 验证步骤。版本锁定确保生产者和消费者使用的atlas-intg库版本与 Atlas Server 完全一致。文档化将你系统中用到的所有消息格式片段整理成内部 Wiki 文档作为团队开发规范。作者署名九师兄专题目录【Apache Atlas】Apache Atlas 资深工程师到专家实战之路目录总目录【目录】技术体系目录注意本文由 AI 辅助生成技术细节请以官方文档为准。生产环境使用前务必充分测试。