Apache Atlas 架構圖

Atlas 支援多資料來源接入:Hive、HBase、Storm等

Type System

Type

Atlas 中定義了一些元資料型別

  1. ── AtlasBaseTypeDef
  2. ├── AtlasEnumDef
  3. └── AtlasStructDef
  4. ├── AtlasBusinessMetadataDef
  5. ├── AtlasClassificationDef
  6. ├── AtlasEntityDef
  7. └── AtlasRelationshipDef
  8. ├── AtlasStructType
  9. ├── AtlasBusinessMetadataType
  10. ├── AtlasClassificationType
  11. ├── AtlasRelationshipType
  12. └── AtlasEntityType
  13. └── AtlasRootEntityType
  14. ├── AtlasType
  15. ├── AtlasArrayType
  16. ├── AtlasBigDecimalType
  17. ├── AtlasBigIntegerType
  18. ├── AtlasByteType
  19. ├── AtlasDateType
  20. ├── AtlasDoubleType
  21. ├── AtlasEnumType
  22. ├── AtlasFloatType
  23. ├── AtlasIntType
  24. ├── AtlasLongType
  25. ├── AtlasMapType
  26. ├── AtlasObjectIdType
  27. ├── AtlasShortType
  28. ├── AtlasStringType
  29. └── AtlasStructType
  30. ├── AtlasBusinessMetadataType
  31. ├── AtlasClassificationType
  32. ├── AtlasEntityType
  33. └── AtlasRelationshipType
  34. ├── AtlasTypeDefStore
  35. └── AtlasTypeDefGraphStore
  36. └── AtlasTypeDefGraphStoreV2
  37. └── StructTypeDefinition
  38. └── HierarchicalTypeDefinition
  39. ├── ClassTypeDefinition
  40. └── TraitTypeDefinition

Entity

Entity 是基於型別的具體實現

  1. AtlasEntity
  2. ├── AtlasEntityExtInfo
  3. ├── AtlasEntitiesWithExtInfo
  4. └── AtlasEntityWithExtInfo
  5. ├── AtlasEntityStore
  6. └── AtlasEntityStoreV2
  7. ├── AtlasEntityStream
  8. └── AtlasEntityStreamForImport
  9. ├── AtlasEntityType
  10. └── AtlasRootEntityType
  11. └── IAtlasEntityChangeNotifier
  12. ├── AtlasEntityChangeNotifier
  13. └── EntityChangeNotifierNop

Attributes

針對模型定義屬性

  1. AtlasAttributeDef
  2. └── AtlasRelationshipAttributeDef

AtlasAttributeDef 屬性欄位:

  1. private String name;
  2. private String typeName;
  3. private boolean isOptional;
  4. private Cardinality cardinality;
  5. private int valuesMinCount;
  6. private int valuesMaxCount;
  7. private boolean isUnique;
  8. private boolean isIndexable;
  9. private boolean includeInNotification;
  10. private String defaultValue;
  11. private String description;
  12. private int searchWeight = DEFAULT_SEARCHWEIGHT;
  13. private IndexType indexType = null;
  14. private List<AtlasConstraintDef> constraints;
  15. private Map<String, String> options;
  16. private String displayName;
  17. 具體實現:
  18. db:
  19. "name": "db",
  20. "typeName": "hive_db",
  21. "isOptional": false,
  22. "isIndexable": true,
  23. "isUnique": false,
  24. "cardinality": "SINGLE"
  25. columns:
  26. "name": "columns",
  27. "typeName": "array<hive_column>",
  28. "isOptional": optional,
  29. "isIndexable": true,
  30. isUnique": false,
  31. "constraints": [ { "type": "ownedRef" } ]
  • isComposite - 是否複合
  • isIndexable - 是否索引
  • isUnique - 是否唯一
  • multiplicity - 指示此屬性是(必需的/可選的/還是可以是多值)的

System specific types and their significance

Referenceable

This type represents all entities that can be searched for using a unique attribute called qualifiedName.

  1. ├── Referenceable
  2. ├── ReferenceableDeserializer
  3. ├── ReferenceableSerializer
  4. └── V1SearchReferenceableSerializer

Hooks

以Hive元資訊採集為例分析採集過程:

全量匯入

import-hive.sh

  1. "${JAVA_BIN}" ${JAVA_PROPERTIES} -cp "${CP}"
  2. org.apache.atlas.hive.bridge.HiveMetaStoreBridge $IMPORT_ARGS
  1. importTables
  2. └── importDatabases [addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java +295]
  3. └── importHiveMetadata [addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java +289]

上面是呼叫過程:

importTables -> importTable --> registerInstances

  1. AtlasEntitiesWithExtInfo ret = null;
  2. EntityMutationResponse response = atlasClientV2.createEntities(entities);
  3. List<AtlasEntityHeader> createdEntities = response.getEntitiesByOperation(EntityMutations.EntityOperation.CREATE);
  4. if (CollectionUtils.isNotEmpty(createdEntities)) {
  5. ret = new AtlasEntitiesWithExtInfo();
  6. for (AtlasEntityHeader createdEntity : createdEntities) {
  7. AtlasEntityWithExtInfo entity = atlasClientV2.getEntityByGuid(createdEntity.getGuid());
  8. ret.addEntity(entity.getEntity());
  9. if (MapUtils.isNotEmpty(entity.getReferredEntities())) {
  10. for (Map.Entry<String, AtlasEntity> entry : entity.getReferredEntities().entrySet()) {
  11. ret.addReferredEntity(entry.getKey(), entry.getValue());
  12. }
  13. }
  14. LOG.info("Created {} entity: name={}, guid={}", entity.getEntity().getTypeName(), entity.getEntity().getAttribute(ATTRIBUTE_QUALIFIED_NAME), entity.getEntity().getGuid());
  15. }
  16. }

通過Http Post 的請求將庫表資料更新至Atlas

atlasClientV2有很多Http介面

Atlas HTTP 客戶端API:

實時監聽

HiveHook implements ExecuteWithHookContext

ExecuteWithHookContext is a new interface that the Pre/Post Execute Hook can run with the HookContext.

實現run()方法來對Hive 相關事件做處理

Hive相關事件:

  1. BaseHiveEvent
  2. ├── AlterTableRename
  3. ├── CreateHiveProcess
  4. ├── DropDatabase
  5. ├── DropTable
  6. ├── CreateDatabase
  7. └── AlterDatabase
  8. └── CreateTable
  9. └── AlterTable
  10. └── AlterTableRenameCol

以create database 為例分析流程:

  1. //處理Hook 上下文資訊
  2. AtlasHiveHookContext context =
  3. new AtlasHiveHookContext(this, oper, hookContext, getKnownObjects(), isSkipTempTables());
  4. //建庫事件處理,提取相關庫資訊
  5. event = new CreateDatabase(context);
  6. if (event != null) {
  7. final UserGroupInformation ugi = hookContext.getUgi() == null ? Utils.getUGI() : hookContext.getUgi();
  8. super.notifyEntities(ActiveEntityFilter.apply(event.getNotificationMessages()), ugi);
  9. }
  10. public enum HookNotificationType {
  11. TYPE_CREATE, TYPE_UPDATE, ENTITY_CREATE, ENTITY_PARTIAL_UPDATE, ENTITY_FULL_UPDATE, ENTITY_DELETE,
  12. ENTITY_CREATE_V2, ENTITY_PARTIAL_UPDATE_V2, ENTITY_FULL_UPDATE_V2, ENTITY_DELETE_V2
  13. }
  14. //操作使用者獲取
  15. if (context.isMetastoreHook()) {
  16. try {
  17. ugi = SecurityUtils.getUGI();
  18. } catch (Exception e) {
  19. //do nothing
  20. }
  21. } else {
  22. ret = getHiveUserName();
  23. if (StringUtils.isEmpty(ret)) {
  24. ugi = getUgi();
  25. }
  26. }
  27. if (ugi != null) {
  28. ret = ugi.getShortUserName();
  29. }
  30. if (StringUtils.isEmpty(ret)) {
  31. try {
  32. ret = UserGroupInformation.getCurrentUser().getShortUserName();
  33. } catch (IOException e) {
  34. LOG.warn("Failed for UserGroupInformation.getCurrentUser() ", e);
  35. ret = System.getProperty("user.name");
  36. }
  37. }

主要:

獲取實體資訊, 傳遞Hook message的型別、操作使用者

notifyEntities 可以看出其他元件HBase、impala也會呼叫該方法進行訊息的傳送

  1. public static void notifyEntities(List<HookNotification> messages, UserGroupInformation ugi, int maxRetries) {
  2. if (executor == null) { // send synchronously
  3. notifyEntitiesInternal(messages, maxRetries, ugi, notificationInterface, logFailedMessages, failedMessagesLogger);
  4. } else {
  5. executor.submit(new Runnable() {
  6. @Override
  7. public void run() {
  8. notifyEntitiesInternal(messages, maxRetries, ugi, notificationInterface, logFailedMessages, failedMessagesLogger);
  9. }
  10. });
  11. }
  12. }

訊息通知框架:

  1. NotificationInterface
  2. ├── AtlasFileSpool
  3. └── AbstractNotification
  4. ├── KafkaNotification
  5. └── Spooler

資料寫入Kafka中:

  1. @Override
  2. public void sendInternal(NotificationType notificationType, List<String> messages) throws NotificationException {
  3. KafkaProducer producer = getOrCreateProducer(notificationType);
  4. sendInternalToProducer(producer, notificationType, messages);
  5. }

根據NotificationType寫入指定topic 中:

  1. private static final Map<NotificationType, String> PRODUCER_TOPIC_MAP = new HashMap<NotificationType, String>() {
  2. {
  3. put(NotificationType.HOOK, ATLAS_HOOK_TOPIC);
  4. put(NotificationType.ENTITIES, ATLAS_ENTITIES_TOPIC);
  5. }
  6. };
  7. NOTIFICATION_HOOK_TOPIC_NAME("atlas.notification.hook.topic.name", "ATLAS_HOOK"),
  8. NOTIFICATION_ENTITIES_TOPIC_NAME("atlas.notification.entities.topic.name", "ATLAS_ENTITIES"),

資料主要寫入兩個Topic中: ATLAS_ENTITIES、ATLAS_HOOK

ATLAS_HOOK是寫入Hook事件訊息, 建立庫的事件元資料資訊會寫入該Topic中

如何唯一確定一個庫:

  1. public String getQualifiedName(Database db) {
  2. return getDatabaseName(db) + QNAME_SEP_METADATA_NAMESPACE + getMetadataNamespace();
  3. }

dbName@clusterName 確定唯一性

外延應用

一個基於Hive hook 實現Impala 元資料重新整理的用例:

AutoRefreshImpala:https://github.com/Observe-secretly/AutoRefreshImpala

參考

[1] Apache Atlas – Data Governance and Metadata framework for Hadoop

[2] Apache Atlas 原始碼