python抽取数据库元数据_0797使用HDP或CDP的Atlas采集CDH6的元数据和血缘

作者&＃xff1a;谢敏灵

Atlas元数据采集框架

目的

基于以上框架&＃xff0c;使用HDP或CDP的Atlas采集CDH6的元数据和血缘&＃xff0c;理论上只需要将相应的Atlas Hook正确部署到CDH6对应的服务上即可。本文以采集Hive元数据和血缘为例&＃xff0c;描述如何部署Atlas Hive Hook到CDH6上。

验证环境

HDP 3.1.5集群。组件版本&＃xff1a;Atlas 2.0.0.3.1.5.0-152&＃xff0c;Hadoop 3.1.1&＃xff0c; Hive 3.1.0&＃xff0c;HBase 2.0.2&＃xff0c;Infra Solr 0.1.0

CDP 7.1.2集群。组件版本&＃xff1a;Atlas 2.0.0.7.1.2.0-96&＃xff0c;Hadoop 3.1.1&＃xff0c; Hive 3.1.3000&＃xff0c;HBase 2.2.3&＃xff0c;Solr 8.4.1

CDH 6.3.2集群。组件版本&＃xff1a;Hive 2.1.1

备注&＃xff1a;为了能在CDH6上采集到字段级别的血缘&＃xff0c;必须确保补丁HIVE-14706已打到CDH6的Hive 2.1.1上。

使用HDP Atlas采集元数据

1.获取Atlas Hive Hook的jar包

在HDP平台上&＃xff0c;通常可以从/usr/hdp/3.1.5.0-152/atlas/hook/hive/atlas-hive-plugin-impl获取Atlas Hive Hook的所有jar包(包括依赖包)&＃xff1a;

2.复制Atlas jar包到CDH节点

将HDP的Atlas Hive Hook的jar包复制到CDH的HS2和HMS节点上。比如&＃xff1a;将jar包复制到/opt/local/hive/lib目录下&＃xff0c;并将ower更改为hive:hive&＃xff1a;

3.复制Atlas配置文件到CDH节点

在CDH的HS2和HMS节点上创建/etc/atlas/conf目录&＃xff0c;并将HDP Atlas配置文件atlas-application.properties复制到该目录&＃xff1a;

Atlas配置文件内容&＃xff1a;

需要将atlas.cluster.name设置为合适的名称&＃xff0c;以便在Atlas Server上区分哪些元数据是来自于该CDH集群。

4.更改CDH的Hive配置

(1)在Hive服务高级配置代码段增加hive.reloadable.aux.jars.path和hive.exec.post.hooks配置项&＃xff1a;

(2)在HiveServer2和Hive Metastore Server的Java配置选项里增加配置&＃xff1a;

重启Hive服务。

5.验证

在CDH的beeline上执行以下SQL&＃xff1a;

create table default.test_hook_hdp_1 (id int, name string);create table default.test_hook_hdp_2 as select * from default.test_hook_hdp_1;

登录HDP的Atlas Web UI&＃xff0c;验证元数据和血缘的采集情况&＃xff1a;

采集到了CDH的default数据库的元数据&＃xff1a;

采集到了CDH的test_hook_hdp_1和test_hook_hdp_2表的元数据&＃xff1a;

采集到了表级血缘&＃xff1a;

采集到了字段级血缘&＃xff1a;

使用CDP Atlas采集元数据

1.获取Atlas Hive Hook的jar包

在CDP平台上&＃xff0c;通常可以从/opt/cloudera/parcels/CDH-7.1.2-1.cdh7.1.2.p0.4253134/lib/hive/lib/atlas-hive-plugin-impl获取Atlas Hive Hook的所有jar包(包括依赖包)&＃xff1a;

2.复制Atlas jar包到CDH节点

将CDP的Atlas Hive Hook的jar包复制到CDH的HS2和HMS节点上。比如&＃xff1a;将jar包复制到/opt/local/hive/lib712目录下&＃xff0c;并将ower更改为hive:hive&＃xff1a;

3.复制Atlas配置文件到CDH节点

在CDH的HS2和HMS节点上创建/etc/atlas/conf目录&＃xff0c;并将CDP Atlas配置文件atlas-application.properties复制到该目录&＃xff1a;

Atlas配置文件内容&＃xff1a;

CDP Atlas增加了一个新的配置参数atlas.metadata.namespace&＃xff0c;用于指定来自某个集群的元数据所在的命名空间&＃xff1b;当然&＃xff0c;也可以继续使用旧的配置参数atlas.cluster.name。CDP Atlas会优先查找atlas.metadata.namespace&＃xff0c;当找不到该参数时&＃xff0c;再查找atlas.cluster.name。

4.更改CDH的Hive配置

(1)在Hive服务高级配置代码段增加hive.reloadable.aux.jars.path和hive.exec.post.hooks配置项&＃xff1a;

(2)在HiveServer2和Hive Metastore Server的Java配置选项里增加配置&＃xff1a;

重启Hive服务。

5.验证

在CDH的beeline上执行&＃xff1a;

create table default.test_hook_cdp_1 (id int, name string);create table default.test_hook_cdp_2 as select * from default.test_hook_cdp_1;

登录CDP的Atlas Web UI&＃xff0c;验证元数据和血缘的采集情况&＃xff1a;

采集到了CDH的default数据库的元数据&＃xff1a;

采集到了CDH的test_hook_cdp_1和test_hook_cdp_2表的元数据&＃xff1a;

采集到了表级血缘&＃xff1a;

采集到了字段级血缘&＃xff1a;

备注&＃xff1a;CDP Atlas Hive Hook的HiveMetaStoreBridge类的getDatabaseName方法使用了Hive3的接口获取database的catalog name&＃xff1a;

public static String getDatabaseName(Database hiveDB){

String dbName      &＃61; hiveDB.getName().toLowerCase();

String catalogName &＃61; hiveDB.getCatalogName() !&＃61; null ? hiveDB.getCatalogName().toLowerCase() : null;

if (StringUtils.isNotEmpty(catalogName) && !StringUtils.equals(catalogName, DEFAULT_METASTORE_CATALOG)) {

dbName &＃61; catalogName &＃43; SEP &＃43; dbName;

}

return dbName;

}

但因为CDH6的Hive是2.1.1版本&＃xff0c;所以&＃xff0c;运行在CDH6环境下的CDP Atlas Hive Hook在获取catalog name时&＃xff0c;会报以下错误&＃xff1a;

java.lang.NoSuchMethodError: org.apache.hadoop.hive.metastore.api.Database.getCatalogName()

针对该问题&＃xff0c;有两种处理方式&＃xff1a;

1.使用HDP Atlas Hive Hook的jar包。因为HDP jar包没有尝试调用Hive3接口获取catalog name(catalog name不作为database name的一部分)&＃xff0c;所以可以规避该错误。

2.修改CDP Atlas Hive Hook的HiveMetaStoreBridge类&＃xff0c;在获取catalog name抛异常时不影响hook的运行&＃xff1a;

public static String getDatabaseName(Database hiveDB){

String dbName      &＃61; hiveDB.getName().toLowerCase();

//String catalogName &＃61; hiveDB.getCatalogName() !&＃61; null ? hiveDB.getCatalogName().toLowerCase() : null;

String catalogName &＃61; null;

try {

if (hiveDB.getCatalogName() !&＃61; null) {

catalogName &＃61; hiveDB.getCatalogName().toLowerCase();

}

} catch (NoSuchMethodError e) {

LOG.warn("Failed while getting catalog name of database");

}

if (StringUtils.isNotEmpty(catalogName) && !StringUtils.equals(catalogName, DEFAULT_METASTORE_CATALOG)) {

dbName &＃61; catalogName &＃43; SEP &＃43; dbName;

}

return dbName;

}