【Atlas】如何扩展 Atlas 的 REST API 添加自定义接口？

Apache Atlas 2.4.0 自定义 REST API 扩展实战从源码到生产部署的完整指南用户问题原文“95. 如何扩展 Atlas 的 REST API 添加自定义接口”本文将深入探讨如何在Apache Atlas 2.4.0中扩展其原生 REST API添加满足特定业务需求的自定义接口。我们将以Kafka Topic 敏感字段自动识别 (kafka_sensitive_topic_discovery)为真实场景手把手教你从零开始通过修改 Atlas 源码、实现业务逻辑、打包部署最终提供一个/api/atlas/v2/custom/kafka/sensitive-topics接口。文章将覆盖完整的开发、测试和验证流程并提供一份可直接用于生产的代码模板。1. 问题引入原生 API 的局限性在某数据平台安全合规团队要求能够一键扫描所有 Kafka Topic并找出其中包含credit_card,ssn,phone_number等敏感字段的 Topic 列表。现有的 Atlas REST API如/api/atlas/v2/search/attribute虽然强大但无法直接表达这种复杂的“字段名模式匹配”逻辑。每次都需要分析师编写复杂的 DSL 查询效率低下且容易出错。最好的解决方案是扩展 Atlas Server增加一个专门的、语义清晰的自定义 API。2. 原理解析Atlas REST API 的架构与扩展点2.1 核心概念JAX-RS 与 Spring Boot 的融合官方/源码解释Atlas 的 Web 层基于Spring Boot构建并使用JAX-RS (Jersey)作为 RESTful Web Services 的实现框架。所有的原生 API 都是通过带有Path,GET,POST等注解的 Java 类Resource来暴露的。通俗类比Atlas Server 就像一栋写字楼每个楼层Package有不同的公司Resource Class。前台Spring Boot DispatcherServlet根据来访者HTTP Request要去的公司名和房间号URL Path将其引导到正确的办公室Method。要增加新服务只需在这栋楼里租一个新的办公室创建新的 Resource Class。技术本质差异写字楼的房间是物理隔离的而 Java 类是在同一个 JVM 进程中共享内存和资源。2.2 关键源码路径与组件入口点:webapp/src/main/java/org/apache/atlas/web/resources包下包含了所有原生的 Resource 类如EntityREST.java,TypesREST.java。核心依赖: 所有 Resource 类都通过 Spring 的Autowired注解注入了核心的服务 Bean如EntityStore,TypeRegistry,DiscoveryService。请求处理流程:HTTP 请求到达 Spring Boot 内嵌的 Tomcat。请求被路由到对应的 JAX-RS Resource Method。Method 调用底层服务如EntityGraphMapper执行业务逻辑。结果被序列化为 JSON 并返回。2.3 扩展方式修改源码 vs. 插件化目前Atlas没有提供官方的插件化机制如 OSGi 或 SPI来动态加载自定义 API。因此唯一可靠的方式是Fork Atlas 源码在其代码库中直接添加新的 Resource 类然后重新编译打包。YesNoClient HTTP RequestSpring Boot DispatcherPath matches /api/atlas/v2/custom/*?CustomKafkaResource.javaNative Resources e.g. EntityREST.javaCall Atlas Core ServicesReturn JSON Response3. 实战步骤开发并部署自定义 Kafka 敏感 Topic 发现 API3.1 步骤一环境准备与源码下载确保已安装Maven 3.6和OpenJDK 11。gitclone https://github.com/apache/atlas.gitcdatlasgitcheckout release-2.4.0# 切换到2.4.0版本3.2 步骤二创建自定义 Resource 类在webapp/src/main/java/org/apache/atlas/web/resources目录下创建新文件CustomKafkaResource.java。// CustomKafkaResource.javapackageorg.apache.atlas.web.resources;importorg.apache.atlas.exception.AtlasBaseException;importorg.apache.atlas.model.discovery.SearchParameters;importorg.apache.atlas.model.discovery.AtlasSearchResult;importorg.apache.atlas.repository.store.graph.v2.EntityGraphMapper;importorg.apache.atlas.type.AtlasTypeRegistry;importorg.springframework.stereotype.Component;importjavax.inject.Inject;importjavax.ws.rs.*;importjavax.ws.rs.core.MediaType;importjava.util.List;importjava.util.ArrayList;importjava.util.Set;importjava.util.HashSet;importjava.util.stream.Collectors;/** * 自定义 REST 资源用于发现包含敏感字段的 Kafka Topic。 */Path(custom/kafka)ComponentpublicclassCustomKafkaResource{privatestaticfinalSetStringSENSITIVE_FIELD_KEYWORDSnewHashSet();static{// 定义敏感字段关键词列表SENSITIVE_FIELD_KEYWORDS.add(credit_card);SENSITIVE_FIELD_KEYWORDS.add(ssn);SENSITIVE_FIELD_KEYWORDS.add(phone_number);SENSITIVE_FIELD_KEYWORDS.add(email);}// 注入 Atlas 核心服务InjectprivateEntityGraphMapperentityGraphMapper;InjectprivateAtlasTypeRegistrytypeRegistry;/** * GET /api/atlas/v2/custom/kafka/sensitive-topics * 返回所有包含敏感字段的 Kafka Topic 列表。 */GETPath(sensitive-topics)Produces({MediaType.APPLICATION_JSON})publicListStringgetSensitiveKafkaTopics()throwsAtlasBaseException{// 1. 构造搜索参数查找所有 kafka_topic 类型的实体SearchParameterssearchParamsnewSearchParameters();searchParams.setTypeName(kafka_topic);searchParams.setLimit(-1);// 获取所有结果// 2. 调用原生的搜索服务AtlasSearchResultsearchResultentityGraphMapper.searchUsingBasic(searchParams);// 3. 过滤出包含敏感字段的 TopicListStringsensitiveTopicsnewArrayList();for(AtlasSearchResult.AtlasEntityHeaderheader:searchResult.getEntities()){// 4. 获取 Topic 的完整实体以便访问其 columns 属性org.apache.atlas.model.instance.AtlasEntityentityentityGraphMapper.getEntityDefinition(header.getGuid());// 5. 检查 columns 属性ObjectcolumnsAttrentity.getAttribute(columns);if(columnsAttrinstanceofList){List?columns(List?)columnsAttr;for(ObjectcolObj:columns){if(colObjinstanceoforg.apache.atlas.model.instance.AtlasEntity.AtlasEntityHeader){StringcolName(String)((org.apache.atlas.model.instance.AtlasEntity.AtlasEntityHeader)colObj).getAttribute(name);// 6. 检查字段名是否包含敏感关键词if(isFieldNameSensitive(colName)){StringtopicQN(String)entity.getAttribute(qualifiedName);sensitiveTopics.add(topicQN);break;// 找到一个即可跳出内层循环}}}}}returnsensitiveTopics;}privatebooleanisFieldNameSensitive(StringfieldName){if(fieldNamenull)returnfalse;StringlowerFieldNamefieldName.toLowerCase();for(Stringkeyword:SENSITIVE_FIELD_KEYWORDS){if(lowerFieldName.contains(keyword)){returntrue;}}returnfalse;}}⚠️重要警告性能考量此示例为了简化一次性加载了所有kafka_topic实体及其列信息。在生产环境中如果 Topic 数量巨大10万这会导致严重的性能问题和内存溢出OOM。必须分页处理或优化查询逻辑。权限控制此 API 继承了 Atlas 的全局认证通过application.properties中的atlas.authentication.method配置但未做细粒度的授权。生产环境中应集成 Ranger 或自定义授权逻辑。3.3 步骤三编译与打包在 Atlas 根目录下执行 Maven 命令进行编译。# 清理并编译跳过测试以加速mvn clean package-DskipTests# 成功后会在 distro/target/ 目录下生成 atlas-2.4.0-server.tar.gzlsdistro/target/atlas-2.4.0-server.tar.gz3.4 步骤四部署与验证备份现有的 Atlas 安装目录。将新生成的atlas-2.4.0-server.tar.gz解压并替换旧的atlas-server目录。重启Atlas Server。验证点调用自定义 API首先确保 Atlas 中已存在一些 Kafka Topic 实体。# 1. 创建一个包含敏感字段的 Kafka Topic 实体curl-uadmin:admin-XPOST-HContent-Type: application/json\-d{ entity: { typeName: kafka_topic, attributes: { qualifiedName: user_events_sensitivekafka-prod, name: user_events_sensitive, columns: [ {typeName:kafka_column, attributes:{name:user_email}}, {typeName:kafka_column, attributes:{name:event_type}} ] } } }http://localhost:21000/api/atlas/v2/entity# 2. 调用自定义 APIcurl-uadmin:admin http://localhost:21000/api/atlas/v2/custom/kafka/sensitive-topics# 验证点输出应为 [user_events_sensitivekafka-prod]4. FAQ 与高级话题FAQQ: 修改源码会不会导致未来升级困难A:会。这是此方案最大的缺点。每次升级 Atlas 版本都需要将自定义代码cherry-pick到新版本的源码中并解决可能的冲突。务必做好代码管理和文档记录。Q: 能否不修改源码通过外部服务调用 Atlas Client 来实现A:可以但不是“扩展 Atlas API”。你可以开发一个独立的 Spring Boot 应用它内部调用 Atlas Java Client (atlas-client-v2) 来实现业务逻辑并暴露自己的 API。这种方式解耦了但失去了“在 Atlas Server 内部直接访问核心服务”的性能优势且需要维护额外的服务。Q: 自定义 API 如何记录审计日志A: Atlas 的审计日志主要由底层服务如EntityGraphMapper触发。你的自定义 API 只要调用了这些服务相关的操作就会被记录。你也可以在自己的方法中显式地调用AuditLog相关的类来记录自定义事件。Q: 如何处理 API 的输入参数和复杂对象A: 使用标准的 JAX-RS 注解。例如QueryParam(keyword) String keyword用于获取 URL 参数Consumes(MediaType.APPLICATION_JSON)配合一个 POJO 类用于接收 JSON Body。Q: 自定义 API 的异常如何处理A: 抛出AtlasBaseException。Atlas 的全局异常处理器 (AtlasExceptionMapper) 会捕获它并将其转换为标准的 JSON 错误响应。监控建议API 调用指标Atlas 内置了对原生 API 的 Metrics通过 Dropwizard Metrics。自定义 API 也会被自动监控。可以通过 JMX 或 Prometheus endpoint (/metrics) 查看org.apache.atlas.web.resources.CustomKafkaResource.getSensitiveKafkaTopics的调用次数、耗时等。错误率监控监控上述指标中的meter和timer的异常计数。生产最佳实践单元测试为你的CustomKafkaResource编写充分的单元测试Mock 掉EntityGraphMapper等依赖。配置化将SENSITIVE_FIELD_KEYWORDS这样的业务规则放到application.properties中通过Value注入方便运维调整。限流对于可能消耗大量资源的 API务必实现限流逻辑防止恶意调用拖垮整个 Atlas Server。作者署名九师兄专题目录【Apache Atlas】Apache Atlas 资深工程师到专家实战之路目录总目录【目录】技术体系目录注意本文由 AI 辅助生成技术细节请以官方文档为准。生产环境使用前务必充分测试。