Great Expectations（GE） によるノートブック型環境（Databricks）でのデータ品質保証方法のまとめ

## 概要
本記事にて、Great Expectations（GE） によるノートブック型環境（Databricks）でのデータ品質保証方法を共有する。GE がツールの概要を提示した上で、コードとその実行結果を提示する。

実行コードとその出力結果を Github に配置してあり、次のリンクで確認が可能。

https://github.com/manabian-/databricks_tecks_for_qiita/blob/main/notebooks/GreatExpectations_on_databricks/GreatExpectations_on_databricks.ipynb

## Great Expectations とは

Great Expectations (GE) とは、データに対する検証、ドキュメント化、および、プロファイリングにより、データ品質の保証と改善を支援する OSS の Python ライブラリである。データに対する品質保証条件（GE では Expectations と呼ぶ）に基づき、データソースへの検証を行い、検証結果をドキュメント化することができる。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9cdea21e-3879-94a4-a9de-28093b877bbe.png)
引用元：[Welcome | Great Expectations](https://docs.greatexpectations.io/docs/)

Expectation を、次のように定義する。下記図では、`passenger_count` というカラムにおける値の最小が1であり、値の最大が6であることを確認している。このような Expectation を複数定義して、データ品質検証を行う。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fe3939df-9625-c7e3-7608-e9ed46c127b2.png)
引用元：[Welcome | Great Expectations](https://docs.greatexpectations.io/docs/)

生成されるドキュメントは、下記の項目が表示される HTML ファイルであり、静的サイトとしてホストすることができる。検証するごとにその時点でのファイルが生成され、過去にさかのぼって確認することも可能。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bb91b651-2f36-1921-e0c5-2c14924ba08f.png)
引用元：[Welcome | Great Expectations](https://docs.greatexpectations.io/docs/)


GE を利用する基本的な手順は次のようになっている。ドキュメントの構成も、手順がベースとなっている。

1. セットアップ
2. データへの接続
3. Expectations の作成
4. データの検証

下記図が、ドキュメントの構成。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5af6b9a2-dc27-e4a9-3cd8-1562f5fcbcbf.png)
引用元：[Welcome | Great Expectations](https://docs.greatexpectations.io/docs/)


検証できるデータソースには次のものがある。本記事後述の手順では、Spark Dataframe、および、Pandas Dataframeを利用している。

- データベース（SQLAlchemy 経由）
- Pandas Dataframe
- Spark Dataframe


SQLAlchemy 経由で接続できるデータベースには、次のものが提示されている。

-   [How to connect to a Athena database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/athena)
-   [How to connect to a BigQuery database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/bigquery)
-   [How to connect to an MSSQL database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/mssql)
-   [How to connect to a MySQL database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/mysql)
-   [How to connect to a PostgreSQL database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/postgres)
-   [How to connect to a Redshift database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/redshift)
-   [How to connect to a Snowflake database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/snowflake)
-   [How to connect to a SQLite database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/sqlite)
-   [How to connect to a Trino database](https://docs.greatexpectations.io/docs/guides/connecting_to_your_data/database/trino) (formerly Presto SQL)


データソースごとに利用できる Expection が異なり、次のリンクが参考になる。

- [Expectation implementations by backend | Great Expectations](https://docs.greatexpectations.io/docs/reference/expectations/implemented_expectations/)
- [Explore Expectations](https://greatexpectations.io/expectations/) 

GE を利用する方法として、CLIによる方法とノートブック型環境による方法がある。本記事後述の手順では後者の方法を実施しており、次のドキュメントが参考となる。

-   [How to instantiate a Data Context without a yml file](https://docs.greatexpectations.io/docs/guides/setup/configuring_data_contexts/how_to_instantiate_a_data_context_without_a_yml_file/)
-   [How to quickly explore Expectations in a notebook | Great Expectations](https://docs.greatexpectations.io/docs/guides/miscellaneous/how_to_quickly_explore_expectations_in_a_notebook/)
-   [How to pass an in-memory DataFrame to a Checkpoint | Great Expectations](https://docs.greatexpectations.io/docs/guides/validation/checkpoints/how_to_pass_an_in_memory_dataframe_to_a_checkpoint/)


データ検証後に、ドキュメントを作成するだけでなく、次のような [Action](https://docs.greatexpectations.io/docs/terms/action) を設定可能。

-   [How to trigger Email as a Validation Action](https://docs.greatexpectations.io/docs/guides/validation/validation_actions/how_to_trigger_email_as_a_validation_action)
-   [How to collect OpenLineage metadata using a Validation Action](https://docs.greatexpectations.io/docs/guides/validation/validation_actions/how_to_collect_openlineage_metadata_using_a_validation_action)
-   [How to trigger Opsgenie notifications as a Validation Action](https://docs.greatexpectations.io/docs/guides/validation/validation_actions/how_to_trigger_opsgenie_notifications_as_a_validation_action)
-   [How to trigger Slack notifications as a Validation Action](https://docs.greatexpectations.io/docs/guides/validation/validation_actions/how_to_trigger_slack_notifications_as_a_validation_action)
-   [How to update Data Docs after validating a Checkpoint](https://docs.greatexpectations.io/docs/guides/validation/validation_actions/how_to_update_data_docs_as_a_validation_action)


GE に関する基本的な情報は、次のリンクに記載されている。

| #    | リンク                                                       | 概要                   |
| ---- | ------------------------------------------------------------ | ---------------------- |
| 1    | [Welcome](https://docs.greatexpectations.io/docs/)           | 概要                   |
| 2    | [Getting started with Great Expectations](https://docs.greatexpectations.io/docs/tutorials/getting_started/tutorial_overview) | チュートリアル         |
| 3    | [Glossary of Terms](https://docs.greatexpectations.io/docs/glossary) | 用語集                 |
| 4    | [Customize your deployment Great Expectations](https://docs.greatexpectations.io/docs/reference/customize_your_deployment) | 利用するための考慮事項 |
| 5    | [Explore Expectations](https://greatexpectations.io/expectations/) | Expectation 一覧       |
| 6    | [Community Page • Great Expectations](https://greatexpectations.io/community) | コミュニティ関連       |
| 7    | [Case studies from Great Expectations](https://greatexpectations.io/case-studies/) | ケーススタディ         |

## ノートブック型環境（Databricks）での Great Expectations の実行

### 1. Greate Exceptions を利用するための事前準備
#### Great Expectations のインストール

```python
%pip install great_expectations -q
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/df785010-cd05-9225-c5a5-7ec7dc0cbe69.png)

#### データ（データフレーム）の準備

```python
schema = '''
`VendorID` INT,
`tpep_pickup_datetime` TIMESTAMP,
`tpep_dropoff_datetime` TIMESTAMP,
`passenger_count` INT,
`trip_distance` DOUBLE,
`RatecodeID` INT,
`store_and_fwd_flag` STRING,
`PULocationID` INT,
`DOLocationID` INT,
`payment_type` INT,
`fare_amount` DOUBLE,
`extra` DOUBLE,
`mta_tax` DOUBLE,
`tip_amount` DOUBLE,
`tolls_amount` DOUBLE,
`improvement_surcharge` DOUBLE,
`total_amount` DOUBLE,
`congestion_surcharge` DOUBLE
'''
 
src_files = [
    "/databricks-datasets/nyctaxi/tripdata/yellow/yellow_tripdata_2019-01.csv.gz",
#     "/databricks-datasets/nyctaxi/tripdata/yellow/yellow_tripdata_2019-02.csv.gz",
]
 
tgt_df = (
    spark
    .read
    .format("csv")
    .schema(schema)
    .option("header", "true")
    .option("inferSchema", "false")
    .load(src_files)
)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dd994781-63b4-a531-10a9-2340d48cae98.png)

### 2. 基本的なデータ品質検証の実施

本章では、次の記事を参考にしている。

- [How to Use Great Expectations in Databricks | Great Expectations](https://docs.greatexpectations.io/docs/deployment_patterns/how_to_use_great_expectations_in_databricks/)

#### Great Expectations のセットアップ

```python
import datetime
 
from ruamel import yaml
 
import great_expectations as ge
from great_expectations.core.batch import RuntimeBatchRequest
from great_expectations.data_context import BaseDataContext
from great_expectations.data_context.types.base import (
    DataContextConfig,
    FilesystemStoreBackendDefaults,
)
```

```python
# root_directory の初期化
dbutils.fs.rm(root_directory_in_spark_api, True)
 
try:
    # ディレクトリを確認
    display(dbutils.fs.ls(root_directory_in_spark_api))
except:
    print('Directory is empty.')
```

```python
# Great expectaions 利用時のエントリーポイントである Data Context を定義
# https://docs.greatexpectations.io/docs/terms/data_context/
 
# great_expectations.yml を参照せずに定義を実施
data_context_config = DataContextConfig(
    store_backend_defaults=FilesystemStoreBackendDefaults(
        root_directory=root_directory
    ),
)
context = BaseDataContext(project_config=data_context_config)
 
# 利用状況の情報共有を提供を停止
# https://docs.greatexpectations.io/docs/reference/anonymous_usage_statistics/
context.anonymous_usage_statistics.enabled = False
```

```python
# ディレクトリを確認
display(dbutils.fs.ls(root_directory_in_spark_api))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e56e9a9d-7e12-9af6-6bf1-a2b777b985b3.png)


#### データへの接続

```python
datasource_name = "taxi_datasource"
dataconnector_name = "databricks_df"
data_asset_name = "nyctaxi_tripdata_yellow_yellow_tripdata"
tgt_deploy_env = "prod"

```

```python
datasource_config = {
    # データソースを定義
    # https://docs.greatexpectations.io/docs/terms/datasource
    "name": datasource_name,
    "class_name": "Datasource",
 
    # execution_engine を定義
    # https://docs.greatexpectations.io/docs/terms/execution_engine/
    "execution_engine": {
        "module_name": "great_expectations.execution_engine",
        "class_name": "SparkDFExecutionEngine",
    },
 
    # データコネクターを定義
    # https://docs.greatexpectations.io/docs/terms/data_connector/
    "data_connectors": {
        dataconnector_name: {
            "module_name": "great_expectations.datasource.data_connector",
            "class_name": "RuntimeDataConnector",
            "batch_identifiers": [
                "some_key_maybe_pipeline_stage",
                "some_other_key_maybe_run_id",
            ],
        }
    },
}
context.add_datasource(**datasource_config)
```

```python
batch_request = RuntimeBatchRequest(
    datasource_name=datasource_name,
    data_connector_name=dataconnector_name,
    data_asset_name = data_asset_name,
    batch_identifiers={
        "some_key_maybe_pipeline_stage": tgt_deploy_env,
        "some_other_key_maybe_run_id": f"my_run_name_{datetime.date.today().strftime('%Y%m%d')}",
    },
    runtime_parameters={"batch_data": tgt_df},
)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8971ef2b-b9ac-3822-8786-e348fed61e03.png)


#### Expectations を作成

```python
checkpoint_config_name = "nyctaxi_tripdata_yellow_yellow_tripdata__checkpoint"
```

```python
# チェックポイントを定義
checkpoint_config = {
    "name":checkpoint_config_name,
    "config_version": 1,
    "class_name": "SimpleCheckpoint",
    "expectation_suite_name": expectation_suite_name,
    "run_name_template": "%Y%m%d-%H%M%S-yctaxi_tripdata_yellow_yellow_tripdata",
}
context.add_checkpoint(**checkpoint_config)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/97d3d3fd-fa46-960c-089a-5a2f9f2d397b.png)


```python
# checkpoints にファイルが作成されたことを確認
checkpoints_file_path = f'{root_directory_in_spark_api}/checkpoints/{checkpoint_config_name}.yml'
print(dbutils.fs.head(checkpoints_file_path))
```

```python

checkpoint_result = context.run_checkpoint(
    checkpoint_name=checkpoint_config_name,
    validations=[
        {
            "batch_request": batch_request,
            "expectation_suite_name": expectation_suite_name,
        }
    ],
)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7c131f13-40dd-06e1-9dbc-3fc497fd8e76.png)


#### 検証結果を確認

```python
# 品質チェック結果を表示
checkpoint_result["success"]
```

```python
# 品質チェック結果の HTML ファイルを表示
first_validation_result_identifier = (
    checkpoint_result.list_validation_result_identifiers()[0]
)
first_run_result = checkpoint_result.run_results[first_validation_result_identifier]
 
docs_path = first_run_result['actions_results']['update_data_docs']['local_site']
 
html = dbutils.fs.head(docs_path,)
 
displayHTML(html)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6d2c3939-b493-9098-9430-b6e56508dc4f.png)


### 3. 品質エラーがある場合の動作検証

```python
# エラーとなる expectation を追加
validator.expect_column_values_to_not_be_null(
    column="congestion_surcharge",
)
 
validator.save_expectation_suite(discard_failed_expectations=False)
```

```python
checkpoint_result = context.run_checkpoint(
    checkpoint_name=checkpoint_config_name,
    validations=[
        {
            "batch_request": batch_request,
            "expectation_suite_name": expectation_suite_name,
        }
    ],
)
```

```python
# 品質チェック結果のを表示
checkpoint_result["success"]
```

```python
# 品質チェック結果の HTML ファイルを表示
first_validation_result_identifier = (
    checkpoint_result.list_validation_result_identifiers()[0]
)
first_run_result = checkpoint_result.run_results[first_validation_result_identifier]
 
docs_path = first_run_result['actions_results']['update_data_docs']['local_site']
 
html = dbutils.fs.head(docs_path,)
 
displayHTML(html)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1e0b95ef-5e10-fe3e-0aa2-5e6d4b12bcf9.png)


### 4. データプロファイリング

本章では、次の記事を参考にしている。

- [Great Expectation on Databricks. Run great_expectations on the hosted… | by Probhakar | Medium](https://probhakar-95.medium.com/great-expectation-on-databricks-8777042e00de)

```python
from great_expectations.profile.basic_dataset_profiler import BasicDatasetProfiler
from great_expectations.dataset.sparkdf_dataset import  SparkDFDataset
from great_expectations.render.renderer import *
from great_expectations.render.view import DefaultJinjaPageView
```

```python
basic_dataset_profiler = BasicDatasetProfiler()
```



```python
from great_expectations.dataset.pandas_dataset import PandasDataset
gdf = PandasDataset(
    tgt_df
    .limit(1000)
    .toPandas()
) 
```

下記のコードにより、spark データフレームでも実行できるか、パフォーマンスに課題あり

```python
from great_expectations.dataset.sparkdf_dataset import SparkDFDataset
gdf = SparkDFDataset(
    tgt_df
    .limit(1000)
) 

print(gdf.spark_df.count())
gdf.spark_df.display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d13c10ed-be36-5a96-8a40-77f58dd599bd.png)


```python
# データを確認
print(gdf.count())
gdf.head()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/529093ec-90e6-fea3-dc18-adfd73047c74.png)


```python
from great_expectations.profile.basic_dataset_profiler import BasicDatasetProfiler
 
# データをプロファイリング
expectation_suite, validation_result = gdf.profile(BasicDatasetProfiler)
```

```python
from great_expectations.render.renderer import (
    ProfilingResultsPageRenderer,
    ExpectationSuitePageRenderer,
)
from great_expectations.render.view import DefaultJinjaPageView
 
profiling_result_document_content = ProfilingResultsPageRenderer().render(validation_result)
expectation_based_on_profiling_document_content = ExpectationSuitePageRenderer().render(expectation_suite)

```

```python
# HTML を生成
profiling_result_HTML = DefaultJinjaPageView().render(profiling_result_document_content) # type string or str
expectation_based_on_profiling_HTML = DefaultJinjaPageView().render(expectation_based_on_profiling_document_content)

displayHTML(profiling_result_HTML)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4ee117ad-e0c4-5df3-d50d-603e5b258921.png)

### 5. リソースのクリーンアップ

```python
dbutils.fs.rm(root_directory_in_spark_api, True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d11fc183-10ed-452a-3c0e-a5739d1afe5a.png)