Griffin数据质量监控工具
为什么要做数据质量监控
garbage in garbage out，处理的数据质量不好，处理完也不会有什么有用的信息
-
数据不一致 企业早期没有进行统一规划设计，大部分信息系统是逐步迭代建设的，系统建设时间长短各异，各系统数据标准也不同。企业业务系统更关注业务层面，各个业务系统均有不同的侧重点，各类数据的属性信息设置和要求不统一。另外，由于各系统的相互独立使用，无法及时同步更新相关信息等各种原因造成各系统间的数据不一致，严重影响了各系统间的数据交互和统一识别，基础数据难以共享利用，数据的深层价值也难以体现。
-
数据不完整 由于企业信息系统的孤立使用，各个业务系统或模块按照各自的需要录入数据，没有统一的录入工具和数据出口，业务系统不需要的信息就不录，造成同样的数据在不同的系统有不同的属性信息，数据完整性无法得到保障。
-
数据不合规 没有统一的数据管理平台和数据源头，数据全生命周期管理不完整，同时企业各信息系统的数据录入环节过于简单且手工参与较多，就数据本身而言，缺少是否重复、合法、对错等校验环节，导致各个信息系统的数据不够准确，格式混乱，各类数据难以集成和统一，没有质量控制导致海量数据因质量过低而难以被利用，且没有相应的数据管理流程。
-
数据不可控 海量数据多头管理，缺少专门对数据管理进行监督和控制的组织。企业各单位和部门关注数据的角度不一样，缺少一个组织从全局的视角对数据进行管理，导致无法建立统一的数据管理标准、流程等，相应的数据管理制度、办法等无法得到落实。同时，企业基础数据质量考核体系也尚未建立，无法保障一系列数据标准、规范、制度、流程得到长效执行。
-
数据冗余 各个信息系统针对数据的标准规范不一、编码规则不一、校验标准不一，且部分业务 系统针对数据的验证标准严重缺失，造成了企业顶层视角的数据出现“一物多码”、“一 码多物”等现象。
数据质量监控方法
- 设计思路
数据质量监控的设计要分为4个模块:数据，规则，告警和反馈
- 数据:需要被监控的数据，可能存放在不同的存储引擎中
- 规则:值如何设计发现异常的规则，一般而言主要是数值的异常和环比等异常监控方式。也会有一些通过算法来发掘异常数据的方法
- 告警:告警是指发告警的动作，这里可以通过微信消息，电话或者短信，邮件
- 反馈:反馈是指对告警内容的反馈，比如说收到的告警内容，要有人员回应该告警消息是否是真的异常，是否需要忽略该异常，是否已经处理了该异常。有了反馈机制，整个数据监控才能形成闭环
- 技术方案
- 最开始可以先关注核心要监控的内容，比如说准确性，那么就对核心的一些指标做监控即可
- 监控平台尽量不要做太复杂的规则逻辑，尽量只对结果数据进行监控。比如要监控日质量是否波动过大，那么把该计算流程提前，先计算好结果表，最后监控平台只监控结果表是否异常即可
- 多数据源。多数据源的监控有两种方式:针对每个数据源定制实现一部分计算逻辑，也可以通过额外的任务将多数据源中的数据结果通过任务写入一个数据源中，再对该数据源进行监控
- 实时数据监控:区别在于扫描周期的不同，因此在设计的时候可以先以离线为主，但是尽量预留好实时监控的设计
Griffin架构
Apache Griffin是一个开源的大数据数据质量解决方案，它支持批处理和流模式两种 数据质量检测方式，可以从不同维度(如离线任务执行完毕后检查源端和目标端的数 据数量是否一致、源表的数据空值数量等)度量数据资产，从而提升数据的准确度、 可信度。
Griffin主要分为Define、Measure和Analyze三个部分:
- Define:主要负责定义数据质量统计的维度，比如数据质量统计的时间跨度、统计的目标(源端和目标端的数据数量是否一致，数据源里某一字段的非空的数量、不重复值的数量、最大值、最小值、top5的值数量等)
- Measure:主要负责执行统计任务，生成统计结果
- Analyze:主要负责保存与展示统计结果
编译安装
- 相关依赖
- JDK (1.8 or later versions)
- MySQL(version 5.6及以上) : 服务元数据
- Hadoop (2.6.0 or later)
- Hive (version 2.x)
- Maven
- Spark (version 2.2.1) :计算批量、实时指标
- Livy(livy-0.5.0-incubating) : 为服务提供 RESTful API 调用 Apache Spark
- ElasticSearch (5.0 or later versions): 存储指标数据
- Spark安装
1、解压缩，设置环境变量 $SPARK_HOME
tar zxvf spark-2.2.1-bin-hadoop2.7.tgz mv spark-2.2.1-bin-hadoop2.7/ /opt/lagou/servers/spark-2.2.1/ # 设置环境变量 vi /etc/profile export SPARK_HOME=/opt/lagou/servers/spark-2.2.1/ export PATH=$PATH:$SPARK_HOME/bin source /etc/profile2、 修改配置文件 $SPARK_HOME/conf/spark-defaults.conf
spark.master yarn spark.eventLog.enabled true spark.eventLog.dir hdfs://centos7-1:9000/spark/logs spark.serializer org.apache.spark.serializer.KryoSerializer spark.yarn.jars hdfs://centos7-1:9000/spark/spark_2.2.1_jars/*备注:上面的路径要创建
拷贝 MySQL 驱动
cp $HIVE_HOME/lib/mysql-connector-java-5.1.46.jar $SPARK_HOME/jars/将 Spark 的 jar 包上传到 hdfs://hadoop1:9000/spark/spark_2.2.1_jars/
hdfs dfs -mkdir -p /spark/logs hdfs dfs -mkdir -p /spark/spark_2.2.1_jars/ hdfs dfs -put /opt/lagou/servers/spark-2.2.1/jars/*.jar /spark/spark_2.2.1_jars/3 、 修改配置文件spark-env.sh
export JAVA_HOME=/usr/java/jdk1.8.0_231/ export HADOOP_HOME=/opt/lagou/servers/hadoop-2.9.2/ export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop export SPARK_DIST_CLASSPATH=$(hadoop classpath) export YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop4、yarn-site.xml 添加配置
<property> <name>yarn.nodemanager.vmem-check-enabled</name> <value>false</value> </property>yarn.nodemanager.vmem-check-enabled:是否检查虚拟内存。 修改所有节点，并重启yarn服务。
不添加该配配置启动spark-shell，有如下错误:Yarn application has already ended! It might have been killed or unable to launch application master.5、测试spark
spark-shellval lines = sc.textFile("/wcinput/wc.txt") lines.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect() -
Livy安装 1、解压缩，设置环境变量 $LIVY_HOME
unzip livy-0.5.0-incubating-bin.zip mv livy-0.5.0-incubating-bin/ ../servers/livy-0.5.0 # 设置环境变量 vi /etc/profile export LIVY_HOME=/opt/lagou/servers/livy-0.5.0 export PATH=$PATH:$LIVY_HOME/bin source /etc/profile2、修改配置文件 conf/livy.conf
livy.server.host = 127.0.0.1 livy.spark.master = yarn livy.spark.deployMode = cluster livy.repl.enable-hive-context = true3、修改配置文件 conf/livy-env.sh
export SPARK_HOME=/opt/lagou/servers/spark-2.2.1 export HADOOP_HOME=/opt/lagou/servers/hadoop-2.9.2/ export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop4、启动服务
cd /opt/lagou/servers/livy-0.5.0 mkdir logs nohup bin/livy-server & - ES安装
1、解压缩
tar zxvf elasticsearch-5.6.0.tar.gz mv elasticsearch-5.6.0/ ../software/2、创建 elasticsearch用户组 及 elasticsearch 用户。不能使用root用户启动ES程 序，需要创建单独的用户去启动ES 服务;
# 创建用户组 groupadd elasticsearch # 创建用户 useradd elasticsearch -g elasticsearch # 修改安装目录的宿主 chown -R elasticsearch:elasticsearch elasticsearch-5.6.0/3、修改linux系统文件 /etc/security/limits.conf
elasticsearch hard nofile 1000000 elasticsearch soft nofile 1000000 * soft nproc 4096 * hard nproc 40964、修改系统文件 /etc/sysctl.conf
# 文件末尾增加: vm.max_map_count=262144 # 执行以下命令，修改才能生效 sysctl -p5、修改es配置文件
/opt/lagou/servers/elasticsearch-5.6.0/config/elasticsearch.ymlnetwork.host: 0.0.0.0/opt/lagou/servers/elasticsearch-5.6.0/config/jvm.options
jvm内存的分配，原来都是2g，修改为1g-Xms1g -Xmx1g6、启动ES服务
# 到ES安装目录下，执行命令(-d表示后台启动) su elasticsearch cd /opt/lagou/servers/elasticsearch-5.6.0/ bin/elasticsearch -d在浏览器中检查:http://localhost:9200/
安装成功的话，可以看到一段json信息
7、在ES里创建griffin索引
curl -XPUT http://localhost:9200/griffin -d ' { "aliases":{ }, "mappings":{ "accuracy":{ "properties":{ "name":{ "fields":{ "keyword":{ "ignore_above":256, "type":"keyword" } }, "type":"text" }, "tmst":{ "type":"date" } } } }, "settings ":{ "index":{ "number_of_replicas":"2", "number_of_shards":"5" } } } ' - Griffin编译准备
1、软件解压缩
cd /opt/lagou/software unzip griffin-griffin-0.5.0.zip mv griffin-griffin-0.5.0/ ../servers/griffin-0.5.0/ cd griffin-0.5.02、在MySQL中创建数据库quartz，并初始化
/opt/lagou/servers/griffin-0.5.0/service/src/main/resources/Init_quartz_mysql_innodb.sql备注:要做简单的修改，主要是增加 use quartz;
# mysql中执行创建数据库 create database quartz; # 命令行执行，创建表 mysql -uhive -p12345678 < Init_quartz_mysql_innodb.sql3、Hadoop和Hive
在HDFS上创建/spark/spark_conf目录，并将Hive的配置文件hive-site.xml上传到该目录下hdfs dfs -mkdir -p /spark/spark_conf hdfs dfs -put $HIVE_HOME/conf/hive-site.xml /spark/spark_conf/备注:将安装 griffin 所在节点上的 hive-site.xml 文件，上传到 HDFS 对应目录中;
4、确保设置以下环境变量(/etc/profile)
export JAVA_HOME=/opt/lagou/servers/hadoop-2.9.2 export SPARK_HOME=/opt/lagou/servers/spark-2.2.1/ export LIVY_HOME=/opt/lagou/servers/livy-0.5.0 export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop -
Griffin编译
1、service/pom.xml文件配置
编辑 service/pom.xml(约113-117行)，增加MySQL JDBC 依赖(即删除注释):<dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>${mysql.java.version}</version> </dependency>2、修改配置文件 service/src/main/resources/application.properties
备注：替换整个文件server.port = 9876 spring.application.name=griffin_service spring.datasource.url=jdbc:mysql://centos7-2:3306/quartz?autoReconnect=true&useSSL=false spring.datasource.username=airflow spring.datasource.password=12345678 spring.jpa.generate-ddl=true spring.datasource.driver-class-name=com.mysql.jdbc.Driver spring.jpa.show-sql=true # Hive metastore hive.metastore.uris=thrift://centos7-3:9083 hive.metastore.dbname=hivemetadata hive.hmshandler.retry.attempts=15 hive.hmshandler.retry.interval=2000ms # Hive cache time cache.evict.hive.fixedRate.in.milliseconds=900000 # Kafka schema registry kafka.schema.registry.url=http://localhost:8081 # Update job instance state at regular intervals jobInstance.fixedDelay.in.milliseconds=60000 # Expired time of job instance which is 7 days that is 604800000 milliseconds.Time unit only supports milliseconds jobInstance.expired.milliseconds=604800000 # schedule predicate job every 5 minutes and repeat 12 times at most #interval time unit s:second m:minute h:hour d:day,only support these four units predicate.job.interval=5m predicate.job.repeat.count=12 # external properties directory location external.config.location= # external BATCH or STREAMING env external.env.location= # login strategy ("default" or "ldap") login.strategy=default # ldap ldap.url=ldap://hostname:port ldap.email=@example.com ldap.searchBase=DC=org,DC=example ldap.searchPattern=(sAMAccountName={0}) # hdfs default name fs.defaultFS= # elasticsearch elasticsearch.host=centos7-2 elasticsearch.port=9200 elasticsearch.scheme=http # elasticsearch.user = user # elasticsearch.password = password # livy livy.uri=http://localhost:8998/batches livy.need.queue=false livy.task.max.concurrent.count=20 livy.task.submit.interval.second=3 livy.task.appId.retry.count=3 # yarn url yarn.uri=http://centos7-3:8088 # griffin event listener internal.event.listeners=GriffinJobEventHook备注:
- 默认端口是8080，为避免和spark端口冲突，这里端口修改为9876
- 需要启动Hive的 metastore 服务
- 如果Griffin、MySQL没有安装在同一节点，请确认用户有权限能够远程登录
3、修改配置文件 service/src/main/resources/quartz.properties
# 将第26行修改为以下内容: org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjob store.StdJDBCDelegate4、修改配置文件 service/src/main/resources/sparkProperties.json sparkProperties.json 在测试环境下使用:
确认file路径是否和配置的一样，还有本机内存是否充足5、修改配置文件 service/src/main/resources/env/env_batch.json
sinks -> hdfs -> config -> path: "hdfs:///griffin/persist" sinks -> ELASTICSEARCH -> config -> api: “"http://localhost:9200/griffin/accuracy”6、编译
cd /opt/lagou/software/griffin-0.5.0 mvn -Dmaven.test.skip=true clean install- 错误处理
遇到SVGElementTagNameMap相关的错误直接找到对应文件，删除对应行就可以
8、jar拷贝
编译完成后，会在service和measure模块的target目录下分别看到 service-0.5.0.jar
和 measure-0.5.0.jar 两个jar，将这两个jar分别拷贝到服务器目录下。# 将 service-0.5.0.jar 拷贝到 /opt/lagou/servers/griffin-0.5.0/ cd /opt/lagou/servers/griffin-0.5.0/service/target cp service-0.5.0.jar /opt/lagou/service/griffin-0.5.0/ # 将 measure-0.5.0.jar 拷贝到 /opt/lagou/servers/griffin- 0.5.0/，并改名 cd /opt/lagou/servers/griffin-0.5.0/measure/target cp measure-0.5.0.jar /opt/lagou/servers/griffin-0.5.0/griffin- measure.jar # 将 griffin-measure.jar 上传到 hdfs:///griffin 中 cd /opt/lagou/servers/griffin-0.5.0 hdfs dfs -mkdir /griffin hdfs dfs -put griffin-measure.jar /griffin备注:spark在yarn集群上执行任务时，需要到HDFS的/griffin目录下加载griffin- measure.jar，避免发生类org.apache.griffin.measure.Application找不到的错误。
- 启动Griffin服务
启动Griffin管理后台:
cd /opt/lagou/servers/griffin-0.5.0 nohup java -jar service-0.5.0.jar>service.out 2>&1 &Apache Griffin的UI:http://localhost:9876
用户名口令:admin / admin
- 简单实用 创建measure，然后创建job调用measure