七、Spark数据血缘

打包spark-atlas-connector

atlas 官方文档中并不支持 spark sql 的解析，需要使用第三方的包。

地址：https://github.com/hortonworks-spark/spark-atlas-connector

一、git clone 后本地进行打包

mvn package -DskipTests

二、打包后在 spark-atlas-connector/spark-atlas-connector-assembly/target 目录有一个 spark-atlas-connector-assembly-${version}.jar 的 jar，将该 jar 上传到服务器。需要注意的是不要上传 spark-atlas-connector/spark-atlas-connector/target 这个目录内的 jar ，缺少相关依赖包

三、将 spark-atlas-connector-assembly-${version}.jar 放到一个固定目录比如/opt/resource

测试spark hook

首先进入spark-sql client

spark-sql --master yarn  \
--jars /opt/resource/spark-atlas-connector_2.11-0.1.0-SNAPSHOT.jar  \
--files /opt/resource/atlas-application.properties \
--conf spark.extraListeners=com.hortonworks.spark.atlas.SparkAtlasEventTracker \
--conf spark.sql.queryExecutionListeners=com.hortonworks.spark.atlas.SparkAtlasEventTracker \
--conf spark.sql.streaming.streamingQueryListeners=com.hortonworks.spark.atlas.SparkAtlasStreamingQueryEventTracker

执行 hera 的一个资源统计任务

CREATE EXTERNAL TABLE  IF NOT EXISTS sucx_hera.ads_hera_task_mem_top_10(
    `job_id` BIGINT COMMENT '任务ID',
    `user` STRING COMMENT '关注人',
    `applicationId` STRING COMMENT 'yarn执行的app id',
    `memorySeconds` BIGINT COMMENT '内存使用时间和',
    `startedTime` BIGINT COMMENT '开始时间',
    `finishedTime` BIGINT COMMENT '结束时间',
    `elapsedTime` BIGINT COMMENT '运行时间',
    `vcoreSeconds` BIGINT COMMENT 'vcore使用时间和')

ROW FORMAT SERDE
    'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH SERDEPROPERTIES (
    'field.delim'='\t',
    'serialization.format'='\t')
STORED AS INPUTFORMAT
    'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
    'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
    'cosn://sucx-big-data/bi//sucx_hera/ads_hera_task_mem_top_10';

insert overwrite table sucx_hera.ads_hera_task_mem_top_10

select
 job_id,user,applicationId,memorySeconds,startedTime,finishedTime,elapsedTime,vcoreSeconds
from
(SELECT
    top.job_id,
    row_number() over(distribute by top.applicationId ORDER BY sso.id) as num,
    case when sso.name is null then operator
    else sso.name end as user,
    top.applicationId,
    top.memorySeconds,
    top.startedTime,
    top.finishedTime,
    top.elapsedTime,
    top.vcoreSeconds

FROM (
select * from sucx_hera.dws_hera_task_mem_top_10 where dt = '20200901' ) top

left join bi_ods.ods_hera_job_monitor monitor

on  monitor.dt='20200901' and top.job_id=monitor.job_id

left join bi_ods.ods_hera_sso sso

on sso.dt='20200901' and find_in_set(sso.id,monitor.user_ids) >0 order by job_id ) temp

where temp.num = 1

执行后，查看 ads_hera_task_mem_top_10 表级血缘

35c3dd0ce089f825c1c9a948e9987d53_640_wx_fmt=png&wxfrom=5&wx_lazy=1&wx_co=1&tp=webp.png

注意此包不支持 spark 字段的支持。

如果需要spark字段的支持，一种是spark代码转成hive跑一遍，一种就是需要自研了。

八、Atlas二次开发

atlas虽好，但是很多场景依然无法满足我们的需要。这时候就不得不做一些改动了。二次开发有两种方式一种是基于Atlas Api的开发，一种是修改源码。

REST API

http://atlas.apache.org/api/v2/index.html
DiscoveryREST
http://hostname:21000/api/atlas/v2/search/basic?classification=class1
参数支持：query、typeName、classification、excludeDeletedEntities、limit、offset
code：https://github.com/apache/atlas/blob/6bacbe946bbc5ca72118304770d5ad920695bd52/webapp/src/main/java/org/apache/atlas/web/rest/DiscoveryREST.java

# 查询所有 Table
http://hostname:21000/api/atlas/v2/search/dsl?typeName=Table

# 查询 owner 前缀为 John 的 Table
http://hostname:21000/api/atlas/v2/search/attribute?typeName=Table&attrName=owner&attrValuePrefix=John

# 查询 Table 的一些属性，如：guid, ownerName, searchParameters 等
http://hostname:21000/api/atlas/v2/search/saved/Table

# 不是很懂（返回的数据和上面那个 API 一模一样）
http://hostname:21000/api/atlas/v2/search/saved

# 查询 EntityType - Table 下有哪些 entity.
http://hostname:21000/api/atlas/v2/search/saved/execute/Table

# 查询 guid 为 e283d8c1-ae19-4f4b-80c0-38031788383b 的 EntityType 下有哪些 entity.
http://hostname:21000/api/atlas/v2/search/saved/execute/guid/e283d8c1-ae19-4f4b-80c0-38031788383b

LineageREST
# 查询 guid 为 a95cb57f-4643-4edf-b5a5-0f1de2e0f076 的实体的血缘
http://hostname:21000/api/atlas/v2/lineage/a95cb57f-4643-4edf-b5a5-0f1de2e0f076

EntityREST
# 查询 guid 为 48f29229-47a9-4b05-b053-91e6484f42a1 的实体
http://hostname:21000/api/atlas/v2/entity/guid/48f29229-47a9-4b05-b053-91e6484f42a1

# 查询 guid 为 48f29229-47a9-4b05-b053-91e6484f42a1 的实体的审计数据
http://hostname:21000/api/atlas/v2/entity/48f29229-47a9-4b05-b053-91e6484f42a1/audit

# 查询 guid 为 48f29229-47a9-4b05-b053-91e6484f42a1 的实体所属的 classifications
http://hostname:21000/api/atlas/v2/entity/guid/48f29229-47a9-4b05-b053-91e6484f42a1/classifications

# 根据 EntityType 的唯一属性查找实体
# 如下：name 是 DB 的一个唯一属性. 查找 name=Logging 的 DB.
http://hostname:21000/api/atlas/v2/entity/uniqueAttribute/type/DB?attr:name=Logging

# 查询 entity 与 classification 是否有关？
http://hostname:21000/api/atlas/v2/entity/guid/48f29229-47a9-4b05-b053-91e6484f42a1/classification/Dimension

# 批量查询实体
http://hostname:21000/api/atlas/v2/entity/bulk?guid=e667f337-8dcc-468b-a5d0-96473f8ede26&guid=a95cb57f-4643-4edf-b5a5-0f1de2e0f076

RelationshipREST
# 查询 guid 为 726c0120-19d2-4978-b38d-b03124033f41 的 relationship
# 注：relationship 可以看做是血缘的一条边
http://hostname:21000/api/atlas/v2/relationship/guid/726c0120-19d2-4978-b38d-b03124033f41

TypesREST
http://hostname:21000/api/atlas/v2/types/typedef/guid/e0ca4c40-6165-4cec-b489-2b8e5fc7112b
http://hostname:21000/api/atlas/v2/types/typedef/name/Table
http://hostname:21000/api/atlas/v2/types/typedefs/headers
http://hostname:21000/api/atlas/v2/types/typedefs
http://hostname:21000/api/atlas/v2/types/enumdef/name/hive_principal_type
http://hostname:21000/api/atlas/v2/types/enumdef/guid/ee30446a-92e1-4bbc-aa0a-66ac21970d88
http://hostname:21000/api/atlas/v2/types/structdef/name/hive_order
http://hostname:21000/api/atlas/v2/types/structdef/guid/0b602605-8c88-4b60-a177-c1c671265294
http://hostname:21000/api/atlas/v2/types/classificationdef/name/PII
http://hostname:21000/api/atlas/v2/types/classificationdef/guid/3992eef8-fd74-4ae7-8b35-fead363b2122
http://hostname:21000/api/atlas/v2/types/entitydef/name/DataSet
http://hostname:21000/api/atlas/v2/types/entitydef/guid/5dca20ce-7d57-4bc3-a3f1-93fa622c32b1
http://hostname:21000/api/atlas/v2/types/relationshipdef/name/process_dataset_outputs
http://hostname:21000/api/atlas/v2/types/relationshipdef/guid/5d76dae0-6bad-4d19-b9b2-cb0cdc3b53d5

GlossaryREST
http://hostname:21000/api/atlas/v2/glossary
http://hostname:21000/api/atlas/v2/glossary/302dc8f8-3bc8-4878-9ba2-0c585ca6de3d

本地开发环境

启动HBase和solr

由于在Atlas需要使用到HBase以及slor。方便调试可以解压之前编译好内置HBase和sole的tar。

文件补充

创建文件夹将，atlas所需要的文件拷贝过去。

将编译内置HBase以及solr源码目录distro/target/conf下的文件拷贝到上述的上图的conf目录。将源码中addons/models下所有的文件拷贝过去。将源码中的atlas/webapp/target中的内容全部拷贝到webapp目录中。

源码启动

将clone下来的源码导入到IDEA中。配置启动参数：

-Datlas.home=/opt/deploy
-Datlas.conf=/opt/deploy/conf
-Datlas.data=/opt/deploy/data

参考文章：

http://atlas.apache.org/

https://blog.csdn.net/su20145104009/article/details/108253556

https://www.cnblogs.com/ttzzyy/p/14143508.html