使用 Airflow 帮助提升数据质量

• 2020 年 2 月 19 日
• 笔记

Airflow 是属于 Apache 基金会的开源项目，可以帮助数据工程师构建完整的数据 ETL（数据抽取、转换和加载）流程。Airflow 使用 DAG（有向无环图）去描述整个工作流的运行流程，使用 Operator 决定工作流中的某个环节该如何执行。

一个 DAG 会包含四五个不同类型的 Operator ，有着两三种文件系统或者数据库的数据转换，例如某个对接方每天会定时把处理好的数据根据省份的数量拆分成 31 个文件传输到 FTP 服务器上，有时候因为对接方资源不足或者是其他什么原因，导致 FTP 服务器上文件出现了缺失，此时运行中的工作流不对文件数量进行检查的话，在后续的 ETL 过程中就会导致数据不准确。因此在整个数据 ETL 过程中，如何证明数据在工作流中是否准确、数据质量能否够高就很重要了。

在实践中，我们认为这个判断或者是证明过程是保证数据质量很重要的部分。因为应用的正确运行需要正确的数据，并且优秀的数据质量可以提高数据使用人员使用数据的信心。我们在思考如何保证数据质量的过程中，发现到纠正结果数据是非常困难的，因此需要尽可能在结果数据的产生过程中确认中间流程没有问题，一旦中间流程的处理的数据出现问题应该及时把任务停止或者是数据没有到齐的话，等待数据到齐再进行下一步流程。

借助于 Airflow 的 Operator 良好的拓展能力，我们基于 BaseOperator 方法扩展了一系列与数据检查相关的 Operator （以下简称 Check Operator），例如检测 FTP 服务器符合给定正则表达式的文件是否存在，数量是否正确。在 Airflow 运行工作流的过程中 Check Operator 会检查数据是否正确，如果数据不正确则会停止整个工作流，反之亦然。

check_sftp_fs = SftpFSDQOperator(task_id='check_sftp_fs',                                   ssh_conn_id='ftp_conn_id',                                   filepath='/path/to/file',                                   file_count=31,                                   dag=dag)  

目前为止，我们使用的检查方式归纳为一句话就是：使用某一种类型的查询条件判断返回的结果是否相等。虽然这种简单粗暴的方式可以避免很多问题，但是我们希望能做到更多，比如针对数据数量的突增突减现象能有一个恰当的反馈，再比如对数据的完整性、规范性、一致性、准确性、唯一性、关联性等方向有一个更好的判断方式。