数据治理核心保障数据质量监控开源项目Apache Griffin分享

概述

定义

Apache Griffin 官网地址 https://griffin.apache.org/ 源码release最新版本0.6.0

Apache Griffin 官网文档地址 https://griffin.apache.org/docs/quickstart.html

Apache Griffin 源码地址 https://github.com/apache/griffin

Apache Griffin是一个开源的大数据数据质量解决方案，它支持批处理和流模式两种数据质量检测方式，可以从不同维度（比如离线任务执行完毕后检查源端和目标端的数据数量是否一致、源表的数据空值数量等）度量数据资产，从而提升数据的准确度、可信度。

Apache Griffin提供了一套定义良好的数据质量领域模型，涵盖了一般情况下的大多数数据质量问题。它还定义了一组数据质量DSL来帮助用户定义他们的质量标准。通过扩展DSL甚至可以在Apache Griffin中实现自定义的特定特性/功能。

数据质量(DQ)是物联网、机器学习等许多数据消费者的关键标准，但如何确定“好”数据没有标准协议。Apache Griffin是一个模型驱动的数据质量服务平台，可以在其中按需检查数据。它提供了一个标准流程来定义数据质量度量、执行和报告，允许跨多个数据系统进行这些检查；当不信任自己的数据或者担心数据会对关键决策产生负面影响时则可以使用Apache Griffin来确保数据质量。

Apache Griffin支持两种类型的数据源：

batch数据：通过数据连接器从Hadoop平台收集数据。
streaming数据：可以连接到诸如Kafka之类的消息系统来做近似实时数据分析。

为何要做数据质量监控

当数据从不同的数据源流向不同的应用系统的时候，缺少端到端的统一视图来追踪数据沿袭(Data Lineage)。这也就导致了在识别和解决数据质量问题上要花费许多不必要的时间。
缺少一个实时的数据质量检测系统。从数据资产(Data Asset)注册，数据质量模型定义，数据质量结果可视化、可监控，当检测到问题时，可以及时发出警报。
缺乏一个共享平台和API服务，让每个项目组无需维护自己的软硬件环境就能解决常见的数据质量问题。

基本概念

DQC：Data Quality Control，数据质量检测/数据质量控制，一般称为数据质量监控。
SLA：Service Level Agreement，也就是服务等级协议，指的是系统服务提供者（Provider）对客户（Costomer）的一个服务承诺，通常称为数据产出分级运维服务。

由定义可知，DQC关注数据口径，负责数据准不准的监测，而SLA关注产出及时性和稳定性，这两者有机结合共同保障了数据质量。在需求场景上DQC主要负责对数据资产质量和波动的监控，SLA主要负责对数据产出和任务调度结果和时长的监控。

特性

度量：精确度、完整性、及时性、唯一性、有效性、一致性。
异常监测：利用预先设定的规则，检测出不符合预期的数据，提供不符合规则数据的下载。
异常告警：通过邮件或门户报告数据质量问题。
可视化监测：利用控制面板来展现数据质量的状态。
实时性：可以实时进行数据质量检测，能够及时发现问题。
可扩展性：可用于多个数据系统仓库的数据校验。
自助服务：Griffin提供了一个简洁易用的用户界面，可以管理数据资产和数据质量规则；同时用户可以通过控制面板查看数据质量结果和自定义显示内容。

架构

Apache Griffin通过3个步骤来处理数据质量问题，步骤如下:

定义数据质量：数据科学家/分析师定义他们的数据质量要求，如准确性、完整性、及时性、唯一性、有效性、一致性和分析等。
测量数据质量：源数据将被摄取到Apache Griffin计算集群中，Apache Griffin将根据数据质量需求启动数据质量测量。
度量结果：作为度量的数据质量报告将被发送到指定的地方。

此外Apache Griffin还为用户提供了一个前端层，用户可以轻松地将任何新的数据质量需求装载到Apache Griffin平台中，并编写全面的逻辑来定义他们的数据质量。

在Griffin的架构中，主要分为Define、Measure和Analyze三个部分

各部分的职责如下：

Define：主要负责定义数据质量统计的维度，比如数据质量统计的时间跨度、统计的目标（源端和目标端的数据数量是否一致，数据源里某一字段的非空的数量、不重复值的数量、最大值、最小值、top5的值数量等）。
Measure：主要负责执行统计任务，生成统计结果。
Analyze：主要负责保存与展示统计结果。

安装

Docker部署

Griffin docker镜像是预先构建在docker hub上的，可以通过docker方式试用体验Apache Griffin。

# 国外地址镜像下载安装
docker pull apachegriffin/griffin_spark2:0.3.0
docker pull apachegriffin/elasticsearch
docker pull apachegriffin/kafka
docker pull zookeeper:3.5
# 中国地址镜像下载安装
docker pull registry.docker-cn.com/apachegriffin/griffin_spark2:0.3.0
docker pull registry.docker-cn.com/apachegriffin/elasticsearch
docker pull registry.docker-cn.com/apachegriffin/kafka
docker pull zookeeper:3.5

docker映像是Apache Griffin环境映像，各镜像包含内容如下：

apachegriffin/griffin_spark2：该镜像包含mysql、hadoop、 hive、 spark、 livy、Apache Griffin服务、Apache Griffin度量，以及一些准备好的demo数据，它作为一个单节点spark集群，提供spark引擎和Apache Griffin服务。
apachegriffin/elasticsearch：此镜像基于官方的elasticsearch，添加了一些配置以启用cors请求，为指标持久化提供elasticsearch服务。
apachegriffin/kafka：此镜像包含kafka 0.8，以及一些演示流数据，以流模式提供流数据源。
zookeeper:3.5：此镜像为官方zookeeper，以流媒体模式提供zookeeper服务。

Docker 镜像批处理使用

下载获取源码中docker/compose/docker-compose-batch.yml文件，Griffin源码目录主要包括griffin-doc、measure、service和ui四个模块
- griffin-doc负责存放Griffin的文档
- measure采用scala语言编写，负责与spark交互，执行统计任务
- service采用java的SpringBoot作为服务实现，负责给ui模块提供交互所需的restful api，保存统计任务，展示统计结果。

通过docker-compose启动

# 启动
docker-compose -f docker-compose-batch.yml up -d
# 查看容器
docker container ls

可以通过使用任何http客户端来尝试Apache Griffin api，这里以postman为例，官方源码中准备了两个postman的json配置文件。

在postman以文件导入上面两个json配置文件，在Griffin Environment配置BASE_PATH环境变量，端口为上面docker容器暴露的38080

先通过调用api (Basic -> Get griffin version)以确保Apache Griffin服务已经启动。

通过api Measures -> Add measure添加一个精度度量，在Apache Griffin中创建一个度量。

通过api jobs -> Add job添加一个作业来调度一个作业来执行度量。在本例中，调度间隔为4分钟，measure.id填写为上一步返回的id值。

几分钟后，可以从elasticsearch获得度量

curl -XGET 'hadoop2:39200/griffin/accuracy/_search?pretty&filter_path=hits.hits._source' -d '{"query":{"match_all":{}},  "sort": [{"tmst": {"order": "asc"}}]}'
{
  "hits" : {
    "hits" : [
      {
        "_source" : {
          "name" : "metricName",
          "tmst" : 1509599811123,
          "value" : {
            "__tmst" : 1509599811123,
            "miss" : 11,
            "total" : 125000,
            "matched" : 124989
          }
        }
      },
      {
        "_source" : {
          "name" : "metricName",
          "tmst" : 1509599811123,
          "value" : {
            "__tmst" : 1509599811123,
            "miss" : 11,
            "total" : 125000,
            "matched" : 124989
          }
        }
      }
    ]
  }
}

Docker 镜像流处理使用

下载获取源码中docker/compose/docker-compose-streaming.yml文件。
通过docker-compose启动

# 启动
docker-compose -f docker-compose-streaming.yml up -d
# 查看容器
docker container ls

执行流测试

# 进入griffin容器
docker exec -it griffin bash
# 切换到measure目录
cd ~/measure
# 执行脚本进行流精度测量
./streaming-accu.sh
# 跟踪日志
tail -f streaming-accu.log

执行流分析测量

# 先杀死上面进行进程
kill -9 `ps -ef | awk '/griffin-measure/{print $2}'`
# 然后清除上次流作业的检查点目录和其他相关目录
./clear.sh
# 执行脚本进行流分析度量
./streaming-prof.sh
# 跟踪日志
tail -f streaming-prof.log

UI界面操作

访问UI http://hadoop2:38080，默认用户密码为griffin/griffin

总体业务流程

当前docker镜像中默认有创建两个数据资产demo_src和demo_tgt可供测试。

# 进入griffin容器
docker exec -it griffin bash
# 进入hive命令行
hive

创建度量标准
- 选择数据源，单一的真实来源与目标进行数据质量比较，目前只能从一个模式中选择属性。
- 选择目标，以便与源进行数据质量比较。
- 将目标数据字段映射到源字段
- 完成demo_src和demo_tgt的分区配置
- 填写度量的必要信息
- 确保度量配置并保存
创建作业来定期处理度量，度量名称选择上面my_accuracy_measure，设置每五分钟执行任务，点击提交按钮确认信息再点击保存按钮