元数据治理利器 - Apache Atlas

一、功能简介

Atlas 是一组可扩展的核心基础治理服务，使企业能够高效地满足其在 Hadoop 中的合规性要求，并允许与整个企业数据生态系统集成。Apache Atlas 为组织提供开放的元数据管理和治理功能，以构建其数据资产的目录，对这些资产进行分类和治理，并为数据科学家、分析师和数据治理团队提供围绕这些数据资产的协作能力。

二、架构设计

Atlas 中主要有以下核心组件

1. Type元数据类型定义，这里可以是数据库，表，列等，还可以细分 mysql 表( mysql_table )，oracle 表( oracle_table )等，atlas自带了很多类型，如 DataSet，Process 等，一般情况下，数据相关的类型在定义类型的时候都会继承 DataSet，而流程相关的类型则会继承 Process，便于生成血缘关系。我们也可以通过调用 api 自定义类型。这是一切的起点，定义完类型之后，才能生成不同类型的元数据实体，生成血缘关系。

2. Classification分类，通俗点就是给元数据打标签，分类是可以传递的，比如 A 视图是基于 A 表生成的，那么如果 A 表打上了 a 这个标签，A 视图也会自动打上 a 标签，这样的好处就是便于数据的追踪。

3. Entity实体，表示具体的元数据，Atlas 管理的对象就是各种 Type 的 Entity。

4. Lineage数据血缘，表示数据之间的传递关系，通过 Lineage 我们可以清晰的知道数据的从何而来又流向何处，中间经历了哪些操作，这样一旦数据出现问题，可以迅速追溯，定位是哪个环节出现错误。

三、服务依赖

在安装Atlas前需要把一下软件都安装好，注意Kafka、Hadoop、Hive、Hbase、Solr都需要注册到同一个ZK下，不可使用自带的ZK。

服务	子服务
Zookeeper	QuorumPeerMain
Kafka	Kafka
MySQL	MySQL
Hadoop	HDFS：NameNode、SecondaryNameNode、DataNode Yarn：NodeManager、ResourceManager
Hive	Hive
Hbase	HMaster、HRegionServer
Solr	Solr
Atlas	Atlas

四、Atlas安装与集成

4.1 解压缩

 tar -zxvf apache-atlas-2.1.0-server.tar.gz

4.2 集成hbase

 vim conf/atlas-application.properties 
 
atlas.graph.storage.hostname=host:2181

 vim conf/atlas-env.sh
 
export HBASE_CONF_DIR=/home/root/apache/hbase-2.0.5/conf

4.3 集成Solr

 vim conf/atlas-application.properties 
 
atlas.graph.index.search.solr.zookeeper-url=host:2181

创建 solr collection

 sudo -i -u root /home/root/apache/solr-7.7.3/bin/solr create -c vertex_index -d  /home/root/apache/atlas/apache-atlas-2.1.0/conf/solr -shards 3 -replicationFactor 2 -force
 
sudo -i -u root /home/root/apache/solr-7.7.3/bin/solr create -c edge_index -d  /home/root/apache/atlas/apache-atlas-2.1.0/conf/solr -shards 3 -replicationFactor 2 -force
 
sudo -i -u root /home/root/apache/solr-7.7.3/bin/solr create -c fulltext_index -d  /home/root/apache/atlas/apache-atlas-2.1.0/conf/solr -shards 3 -replicationFactor 2 -force

创建好后可以在页面上看到

4.4 集成kafka

 vim conf/atlas-application.properties 
 
atlas.notification.embedded=false
atlas.kafka.data=/home/root/data/kafka
atlas.kafka.zookeeper.connect=host:2181
atlas.kafka.bootstrap.servers=host:9092

4.5 Atlas服务配置

 vim conf/atlas-application.properties 
 
atlas.rest.address=host:21000
atlas.server.run.setup.on.start=false
tlas.audit.hbase.zookeeper.quorum=host:2181

4.6 集成hive

修改atlas-application.properties

 vim conf/atlas-application.properties 
 
atlas.hook.hive.synchronous=false
atlas.hook.hive.numRetries=3
atlas.hook.hive.queueSize=10000
atlas.cluster.name=primary

修改hive-site.xml

 vim hive-3.1.2/conf/hive-site.xml  
 
hive.exec.post.hooks
org.apache.atlas.hive.hook.HiveHook

安装hive-hook

 tar -zxvf apache-atlas-2.1.0-hive-hook.tar.gz 
 
cp -r apache-atlas-hive-hook-2.1.0/* apache-atlas-2.1.0/

修改hive-env.sh

 vim hive-3.1.2/conf/hive-env.sh 
 
export HIVE_AUX_JARS_PATH=/home/root/apache/atlas/apache-atlas-2.1.0/hook/hive

拷贝atlas-application.properties到hive conf目录

 cp atlas-application.properties ../../../hive-3.1.2/conf/

五、Atlas使用

5.1 Atlas启动

 bin/atlas_start.py

访问host:21000，账号和密码都是admin

5.2 初始化元数据

 cd hook-bin
sh import-hive.sh

5.3 增量同步

输入hive命令进入控制台，创建订单事实表

 CREATE TABLE dwd_order_info (
`id` STRING COMMENT '订单号',
`final_amount` DECIMAL(16,2) COMMENT '订单最终金额',
`order_status` STRING COMMENT '订单状态',
`user_id` STRING COMMENT '用户 id',
`payment_way` STRING COMMENT '支付方式',
`delivery_address` STRING COMMENT '送货地址',
`out_trade_no` STRING COMMENT '支付流水号',
`create_time` STRING COMMENT '创建时间',
`operate_time` STRING COMMENT '操作时间',
`expire_time` STRING COMMENT '过期时间',
`tracking_no` STRING COMMENT '物流单编号',
`province_id` STRING COMMENT '省份 ID',
`activity_reduce_amount` DECIMAL(16,2) COMMENT '活动减免金额',
`coupon_reduce_amount` DECIMAL(16,2) COMMENT '优惠券减免金额',
`original_amount` DECIMAL(16,2) COMMENT '订单原价金额',
`feight_fee` DECIMAL(16,2) COMMENT '运费',
`feight_fee_reduce` DECIMAL(16,2) COMMENT '运费减免'
) COMMENT '订单表'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

创建地域维表

 CREATE TABLE dim_base_province (
`id` STRING COMMENT '编号',
`name` STRING COMMENT '省份名称',
`region_id` STRING COMMENT '地区 ID',
`area_code` STRING COMMENT '地区编码',
`iso_code` STRING COMMENT 'ISO-3166 编码，供可视化使用',
`iso_3166_2` STRING COMMENT 'IOS-3166-2 编码，供可视化使用'
) COMMENT '省份表'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

上传测试数据

 cd /home/root/apache/hive-3.1.2/test-data
 
hdfs dfs -put base_province.txt /home/root/data/hive/warehouse/my_db.db/dim_base_province
 
hdfs dfs -put order_info.txt /home/root/data/hive/warehouse/my_db.db/dwd_order_info

创建需求表，根据订单事实表和地区维度表，求出每个省份的订单次数和订单金额

 CREATE TABLE `ads_order_by_province` (
`dt` STRING COMMENT '统计日期',
`province_id` STRING COMMENT '省份 id',
`province_name` STRING COMMENT '省份名称',
`area_code` STRING COMMENT '地区编码',
`iso_code` STRING COMMENT '国际标准地区编码',
`iso_code_3166_2` STRING COMMENT '国际标准地区编码',
`order_count` BIGINT COMMENT '订单数',
`order_amount` DECIMAL(16,2) COMMENT '订单金额'
) COMMENT '各省份订单统计'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

执行sql导入

 set hive.execution.engine=spark;
 
insert into table ads_order_by_province
select
'2021-08-30' dt,
bp.id,
bp.name,
bp.area_code,
bp.iso_code,
bp.iso_3166_2,
count(*) order_count,
sum(oi.final_amount) order_amount
from dwd_order_info oi
left join dim_base_province bp
on oi.province_id=bp.id
group by bp.id,bp.name,bp.area_code,bp.iso_code,bp.iso_3166_2;

再导入到拷贝表一份

 CREATE TABLE `ads_order_by_province_copy` (
`dt` STRING COMMENT '统计日期',
`province_id` STRING COMMENT '省份 id',
`province_name` STRING COMMENT '省份名称',
`area_code` STRING COMMENT '地区编码',
`iso_code` STRING COMMENT '国际标准地区编码',
`iso_code_3166_2` STRING COMMENT '国际标准地区编码',
`order_count` BIGINT COMMENT '订单数',
`order_amount` DECIMAL(16,2) COMMENT '订单金额'
) COMMENT '各省份订单统计'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';
 
set hive.execution.engine=spark;
 
insert into table ads_order_by_province_copy
select * from ads_order_by_province;

5.4 血缘查询结果

我们可以看到，经过上面的sql操作，Atlas已经记录好了血缘关系

六、工作原理

其实就是通过hive-hook监听数仓数据变化，把监听信息发Kafka : ATLAS_HOOK，消息体就是对Atlas数据模型的定义

Atlas读取到消息后会生成数据模型，存储到 Hbase 中，生成表与表，列与列的血缘关系依赖，在前台展示给用户查看。

参考资料

https://atlas.apache.org/#/

https://blog.csdn.net/m0_37719874/article/details/124245209
https://www.bilibili.com/video/BV1jA411F76d?spm_id_from=333.337.search-card.all.click

相关阅读:
麒麟信安操作系统衍生产品解决方案 | 录屏审计软件，让用户桌面操作行为更合规！
剑指 Offer 05. 替换空格
 通过串口实现printf函数，中断实现串口数据接收
 -凯撒密码-
【macOS】Win通过VNC远程控制Macbook
vue.js毕业设计，基于vue.js前后端分离订座预约系统设计与实现（H5移动项目）
Linux常用命令
 Flask Web——表单
 爬虫学习的一个综合案例——访问网站
 Zookeeper 源码分析流程
原文地址：https://www.cnblogs.com/wwzyy/p/16591908.html

	vim conf/atlas-application.properties

	atlas.graph.storage.hostname=host:2181

	vim conf/atlas-env.sh

	export HBASE_CONF_DIR=/home/root/apache/hbase-2.0.5/conf

	vim conf/atlas-application.properties

	atlas.graph.index.search.solr.zookeeper-url=host:2181

	sudo -i -u root /home/root/apache/solr-7.7.3/bin/solr create -c vertex_index -d /home/root/apache/atlas/apache-atlas-2.1.0/conf/solr -shards 3 -replicationFactor 2 -force

	sudo -i -u root /home/root/apache/solr-7.7.3/bin/solr create -c edge_index -d /home/root/apache/atlas/apache-atlas-2.1.0/conf/solr -shards 3 -replicationFactor 2 -force

	sudo -i -u root /home/root/apache/solr-7.7.3/bin/solr create -c fulltext_index -d /home/root/apache/atlas/apache-atlas-2.1.0/conf/solr -shards 3 -replicationFactor 2 -force

	vim conf/atlas-application.properties

	atlas.notification.embedded=false
	atlas.kafka.data=/home/root/data/kafka
	atlas.kafka.zookeeper.connect=host:2181
	atlas.kafka.bootstrap.servers=host:9092

	vim conf/atlas-application.properties

	atlas.rest.address=host:21000
	atlas.server.run.setup.on.start=false
	tlas.audit.hbase.zookeeper.quorum=host:2181

	vim conf/atlas-application.properties

	atlas.hook.hive.synchronous=false
	atlas.hook.hive.numRetries=3
	atlas.hook.hive.queueSize=10000
	atlas.cluster.name=primary

	vim hive-3.1.2/conf/hive-site.xml


	hive.exec.post.hooks
	org.apache.atlas.hive.hook.HiveHook

	tar -zxvf apache-atlas-2.1.0-hive-hook.tar.gz

	cp -r apache-atlas-hive-hook-2.1.0/* apache-atlas-2.1.0/

	vim hive-3.1.2/conf/hive-env.sh

	export HIVE_AUX_JARS_PATH=/home/root/apache/atlas/apache-atlas-2.1.0/hook/hive

	CREATE TABLE dwd_order_info (
	`id` STRING COMMENT '订单号',
	`final_amount` DECIMAL(16,2) COMMENT '订单最终金额',
	`order_status` STRING COMMENT '订单状态',
	`user_id` STRING COMMENT '用户 id',
	`payment_way` STRING COMMENT '支付方式',
	`delivery_address` STRING COMMENT '送货地址',
	`out_trade_no` STRING COMMENT '支付流水号',
	`create_time` STRING COMMENT '创建时间',
	`operate_time` STRING COMMENT '操作时间',
	`expire_time` STRING COMMENT '过期时间',
	`tracking_no` STRING COMMENT '物流单编号',
	`province_id` STRING COMMENT '省份 ID',
	`activity_reduce_amount` DECIMAL(16,2) COMMENT '活动减免金额',
	`coupon_reduce_amount` DECIMAL(16,2) COMMENT '优惠券减免金额',
	`original_amount` DECIMAL(16,2) COMMENT '订单原价金额',
	`feight_fee` DECIMAL(16,2) COMMENT '运费',
	`feight_fee_reduce` DECIMAL(16,2) COMMENT '运费减免'
	) COMMENT '订单表'
	ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

	CREATE TABLE dim_base_province (
	`id` STRING COMMENT '编号',
	`name` STRING COMMENT '省份名称',
	`region_id` STRING COMMENT '地区 ID',
	`area_code` STRING COMMENT '地区编码',
	`iso_code` STRING COMMENT 'ISO-3166 编码，供可视化使用',
	`iso_3166_2` STRING COMMENT 'IOS-3166-2 编码，供可视化使用'
	) COMMENT '省份表'
	ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

	cd /home/root/apache/hive-3.1.2/test-data

	hdfs dfs -put base_province.txt /home/root/data/hive/warehouse/my_db.db/dim_base_province

	hdfs dfs -put order_info.txt /home/root/data/hive/warehouse/my_db.db/dwd_order_info

	CREATE TABLE `ads_order_by_province` (
	`dt` STRING COMMENT '统计日期',
	`province_id` STRING COMMENT '省份 id',
	`province_name` STRING COMMENT '省份名称',
	`area_code` STRING COMMENT '地区编码',
	`iso_code` STRING COMMENT '国际标准地区编码',
	`iso_code_3166_2` STRING COMMENT '国际标准地区编码',
	`order_count` BIGINT COMMENT '订单数',
	`order_amount` DECIMAL(16,2) COMMENT '订单金额'
	) COMMENT '各省份订单统计'
	ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

	set hive.execution.engine=spark;

	insert into table ads_order_by_province
	select
	'2021-08-30' dt,
	bp.id,
	bp.name,
	bp.area_code,
	bp.iso_code,
	bp.iso_3166_2,
	count(*) order_count,
	sum(oi.final_amount) order_amount
	from dwd_order_info oi
	left join dim_base_province bp
	on oi.province_id=bp.id
	group by bp.id,bp.name,bp.area_code,bp.iso_code,bp.iso_3166_2;

	CREATE TABLE `ads_order_by_province_copy` (
	`dt` STRING COMMENT '统计日期',
	`province_id` STRING COMMENT '省份 id',
	`province_name` STRING COMMENT '省份名称',
	`area_code` STRING COMMENT '地区编码',
	`iso_code` STRING COMMENT '国际标准地区编码',
	`iso_code_3166_2` STRING COMMENT '国际标准地区编码',
	`order_count` BIGINT COMMENT '订单数',
	`order_amount` DECIMAL(16,2) COMMENT '订单金额'
	) COMMENT '各省份订单统计'
	ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

	set hive.execution.engine=spark;

	insert into table ads_order_by_province_copy
	select * from ads_order_by_province;