[Apache Atlas] Atlas 架構設計及源程式碼簡單分析

2021 年 9 月 18 日
筆記
[Apache Atlas], Apache Atlas, Atlas, Hive元數據, 數據治理, 數據血緣

Apache Atlas 架構圖

Atlas 支援多數據源接入:Hive、HBase、Storm等

Type System

Type

Atlas 中定義了一些元數據類型

── AtlasBaseTypeDef
  │   ├── AtlasEnumDef
  │   └── AtlasStructDef
  │       ├── AtlasBusinessMetadataDef
  │       ├── AtlasClassificationDef
  │       ├── AtlasEntityDef
  │       └── AtlasRelationshipDef
  ├── AtlasStructType
  │   ├── AtlasBusinessMetadataType
  │   ├── AtlasClassificationType
  │   ├── AtlasRelationshipType
  │   └── AtlasEntityType
  │       └── AtlasRootEntityType
  ├── AtlasType
  │   ├── AtlasArrayType
  │   ├── AtlasBigDecimalType
  │   ├── AtlasBigIntegerType
  │   ├── AtlasByteType
  │   ├── AtlasDateType
  │   ├── AtlasDoubleType
  │   ├── AtlasEnumType
  │   ├── AtlasFloatType
  │   ├── AtlasIntType
  │   ├── AtlasLongType
  │   ├── AtlasMapType
  │   ├── AtlasObjectIdType
  │   ├── AtlasShortType
  │   ├── AtlasStringType
  │   └── AtlasStructType
  │       ├── AtlasBusinessMetadataType
  │       ├── AtlasClassificationType
  │       ├── AtlasEntityType
  │       └── AtlasRelationshipType
  ├── AtlasTypeDefStore
  │   └── AtlasTypeDefGraphStore
  │       └── AtlasTypeDefGraphStoreV2
  └── StructTypeDefinition
      └── HierarchicalTypeDefinition
          ├── ClassTypeDefinition
          └── TraitTypeDefinition

Entity

Entity 是基於類型的具體實現

AtlasEntity
  ├── AtlasEntityExtInfo
  │   ├── AtlasEntitiesWithExtInfo
  │   └── AtlasEntityWithExtInfo
  ├── AtlasEntityStore
  │   └── AtlasEntityStoreV2
  ├── AtlasEntityStream
  │   └── AtlasEntityStreamForImport
  ├── AtlasEntityType
  │   └── AtlasRootEntityType
  └── IAtlasEntityChangeNotifier
      ├── AtlasEntityChangeNotifier
      └── EntityChangeNotifierNop

Attributes

針對模型定義屬性

AtlasAttributeDef
      └── AtlasRelationshipAttributeDef

AtlasAttributeDef 屬性欄位:

private String                   name;
private String                   typeName;
private boolean                  isOptional;
private Cardinality              cardinality;
private int                      valuesMinCount;
private int                      valuesMaxCount;
private boolean                  isUnique;
private boolean                  isIndexable;
private boolean                  includeInNotification;
private String                   defaultValue;
private String                   description;
private int                      searchWeight = DEFAULT_SEARCHWEIGHT;
private IndexType                indexType    = null;
private List<AtlasConstraintDef> constraints;
private Map<String, String>      options;
private String                   displayName;

具體實現：

db:
    "name":        "db",
    "typeName":    "hive_db",
    "isOptional":  false,
    "isIndexable": true,
    "isUnique":    false,
    "cardinality": "SINGLE"
    
    
columns:
    "name":        "columns",
    "typeName":    "array<hive_column>",
    "isOptional":  optional,
    "isIndexable": true,
    「isUnique":    false,
    "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.

  ├── Referenceable
  ├── ReferenceableDeserializer
  ├── ReferenceableSerializer
  └── V1SearchReferenceableSerializer

Hooks

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

全量導入

import-hive.sh

"${JAVA_BIN}" ${JAVA_PROPERTIES} -cp "${CP}" 
org.apache.atlas.hive.bridge.HiveMetaStoreBridge $IMPORT_ARGS

importTables
  └── importDatabases        [addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java +295]
      └── importHiveMetadata [addons/hive-bridge/src/main/java/org/apache/atlas/hive/bridge/HiveMetaStoreBridge.java +289]

上面是調用過程：

importTables -> importTable –> registerInstances

AtlasEntitiesWithExtInfo ret = null;
EntityMutationResponse   response        = atlasClientV2.createEntities(entities);
List<AtlasEntityHeader>  createdEntities = response.getEntitiesByOperation(EntityMutations.EntityOperation.CREATE);

if (CollectionUtils.isNotEmpty(createdEntities)) {
    ret = new AtlasEntitiesWithExtInfo();

    for (AtlasEntityHeader createdEntity : createdEntities) {
        AtlasEntityWithExtInfo entity = atlasClientV2.getEntityByGuid(createdEntity.getGuid());

        ret.addEntity(entity.getEntity());

        if (MapUtils.isNotEmpty(entity.getReferredEntities())) {
            for (Map.Entry<String, AtlasEntity> entry : entity.getReferredEntities().entrySet()) {
                ret.addReferredEntity(entry.getKey(), entry.getValue());
            }
        }

        LOG.info("Created {} entity: name={}, guid={}", entity.getEntity().getTypeName(), entity.getEntity().getAttribute(ATTRIBUTE_QUALIFIED_NAME), entity.getEntity().getGuid());
    }
}

通過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相關事件:

BaseHiveEvent
      ├── AlterTableRename
      ├── CreateHiveProcess
      ├── DropDatabase
      ├── DropTable
      ├── CreateDatabase
      │   └── AlterDatabase
      └── CreateTable
          └── AlterTable
              └── AlterTableRenameCol

以create database 為例分析流程:

//處理Hook 上下文資訊
AtlasHiveHookContext context = 
new AtlasHiveHookContext(this, oper, hookContext, getKnownObjects(), isSkipTempTables());

//建庫事件處理,提取相關庫資訊
event = new CreateDatabase(context);

if (event != null) {
    final UserGroupInformation ugi = hookContext.getUgi() == null ? Utils.getUGI() : hookContext.getUgi();
    super.notifyEntities(ActiveEntityFilter.apply(event.getNotificationMessages()), ugi);
}


public enum HookNotificationType {
    TYPE_CREATE, TYPE_UPDATE, ENTITY_CREATE, ENTITY_PARTIAL_UPDATE, ENTITY_FULL_UPDATE, ENTITY_DELETE,
    ENTITY_CREATE_V2, ENTITY_PARTIAL_UPDATE_V2, ENTITY_FULL_UPDATE_V2, ENTITY_DELETE_V2
}

//操作用戶獲取
if (context.isMetastoreHook()) {
    try {
        ugi = SecurityUtils.getUGI();
    } catch (Exception e) {
        //do nothing
    }
} else {
    ret = getHiveUserName();

    if (StringUtils.isEmpty(ret)) {
        ugi = getUgi();
    }
}

if (ugi != null) {
    ret = ugi.getShortUserName();
}

if (StringUtils.isEmpty(ret)) {
    try {
        ret = UserGroupInformation.getCurrentUser().getShortUserName();
    } catch (IOException e) {
        LOG.warn("Failed for UserGroupInformation.getCurrentUser() ", e);

        ret = System.getProperty("user.name");
    }
}

主要:

獲取實體資訊, 傳遞Hook message的類型、操作用戶

notifyEntities 可以看出其他組件HBase、impala也會調用該方法進行消息的發送

public static void notifyEntities(List<HookNotification> messages, UserGroupInformation ugi, int maxRetries) {
    if (executor == null) { // send synchronously
        notifyEntitiesInternal(messages, maxRetries, ugi, notificationInterface, logFailedMessages, failedMessagesLogger);
    } else {
        executor.submit(new Runnable() {
            @Override
            public void run() {
                notifyEntitiesInternal(messages, maxRetries, ugi, notificationInterface, logFailedMessages, failedMessagesLogger);
            }
        });
    }
}

消息通知框架:

NotificationInterface
      ├── AtlasFileSpool
      └── AbstractNotification
          ├── KafkaNotification
          └── Spooler

數據寫入Kafka中:

@Override
public void sendInternal(NotificationType notificationType, List<String> messages) throws NotificationException {
    KafkaProducer producer = getOrCreateProducer(notificationType);

    sendInternalToProducer(producer, notificationType, messages);
}

根據NotificationType寫入指定topic 中:

private static final Map<NotificationType, String> PRODUCER_TOPIC_MAP = new HashMap<NotificationType, String>() {
    {
        put(NotificationType.HOOK, ATLAS_HOOK_TOPIC);
        put(NotificationType.ENTITIES, ATLAS_ENTITIES_TOPIC);
    }
};

NOTIFICATION_HOOK_TOPIC_NAME("atlas.notification.hook.topic.name", "ATLAS_HOOK"),
NOTIFICATION_ENTITIES_TOPIC_NAME("atlas.notification.entities.topic.name", "ATLAS_ENTITIES"),

數據主要寫入兩個Topic中: ATLAS_ENTITIES、ATLAS_HOOK

ATLAS_HOOK是寫入Hook事件消息, 創建庫的事件元數據資訊會寫入該Topic中

如何唯一確定一個庫：

public String getQualifiedName(Database db) {
    return getDatabaseName(db) + QNAME_SEP_METADATA_NAMESPACE + getMetadataNamespace();
}

dbName@clusterName 確定唯一性

外延應用

一個基於Hive hook 實現Impala 元數據刷新的用例:
AutoRefreshImpala：//github.com/Observe-secretly/AutoRefreshImpala

參考

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

Tags: [Apache Atlas] Apache Atlas Atlas Hive元數據數據治理數據血緣