Tracks
データエンジニア Pythonで
40時間
Great Expectationsのセットアップ
このチュートリアルでは、GX Core（Great Expectationsのオープンソース版）を使って、PandasのDataFrameを検証する方法を学びます。コンテキストのセットアップ、Pandasデータソースの登録、エクスペクテーションの定義、データバッチの検証までを順を追って解説します。
注: DataLabノートブックに沿って進めることをおすすめしますが、独自のPythonスクリプトを作成しても構いません。
1. Great Expectationsのインストール
前提条件
- Python 3.9〜3.12 がインストールされていること。
- 競合を避けるため、仮想環境内にGreat Expectationsをインストールすることを強く推奨します（免責事項: 仮想環境の構築は本記事の範囲外です）。
- サンプルデータセット。
注: 提供されているDataLabノートブックを使用する場合、これらの前提条件は満たされています。スキップして構いません。
以下のコマンドでpip経由でGXをインストールします。
pip install great_expectationsこのコマンドでコアパッケージと必要な依存関係がすべてインストールされます。
2. データコンテキストの初期化
Great Expectationsは構成を管理するためにデータコンテキストを必要とします。ここでは構成を永続化しないために、一時的なデータコンテキストを使用します。
import great_expectations as gx
# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"はじめてのデータ検証スイートを作成する
GXのセットアップができたので、データ検証スイートを作成しましょう。
1. データソースに接続してデータアセットを作成する
データソースはGreat Expectationsをデータに接続するためのもので、データアセットは特定のデータの集合（例：テーブル、DataFrame、ファイル）を表します。
ここでは、inventory_parts_dfというDataFrameに接続する準備を行います。サンプルデータセットは提供のDataLabに含まれており、SQLブロックを実行すると作成されます。
DataLabを使用しない場合は、サンプルデータで独自にDataFrameを作成してください。
では、データソースとアセットを作成します。
# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")2. バッチ定義の追加
バッチ定義は、検証対象のデータを識別・整理するものです。ここでは、DataFrame全体を対象とするバッチ定義を追加します。
# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name3. バッチの取得
バッチはバッチ定義に紐づくデータの集合です。データを検証するには、バッチを取得して、この場合はinventory_parts_dfのDataFrameにリンクする必要があります。
# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)4. スイートの作成とエクスペクテーションの定義
エクスペクテーションはデータ検証のためのルールです。この例では、次のシンプルなエクスペクテーションを定義します。
inventory_idの値がNULLでないこと。part_numの値が一意であること。
# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)利用可能なエクスペクテーションはExpectation Galleryで確認できます。いくつか追加してみてください。
エクスペクテーションを定義すると、GXはエクスペクテーションスイートの構成を出力します。
{
"name": "inventory_parts_suite",
"id": "b2de0b69-0869-4163-8dde-6c09884483f7",
"expectations": [
{
"type": "expect_column_values_to_not_be_null",
"kwargs": {
"column": "inventory_id"
},
"meta": {},
"id": "53d6c42a-d190-412f-a113-783b706531f4"
},
{
"type": "expect_column_values_to_be_unique",
"kwargs": {
"column": "part_num"
},
"meta": {},
"id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
}
],
"meta": {
"great_expectations_version": "1.2.4"
},
"notes": null
}このスイートには次の情報が含まれます。
- スイート名とID： ユニークな名前（
inventory_parts_suite）と識別子で、スイートの追跡と管理に使用します。 - エクスペクテーション： 各ルールには以下が含まれます。
- チェックの種類（例：列にNULLがない、一意であるなど）。
- 検証対象の列などのパラメータ。
- メタデータと各エクスペクテーションの一意のID。追跡やカスタマイズが容易になります。
- メタデータ： Great Expectationsのバージョン情報。ツールとの互換性を担保します。
- ノート： スイートに関する説明的なコメントを追加するためのプレースホルダー（任意）。
この構造化された出力は、データセットを検証するためのドキュメントであると同時に再利用可能な構成として機能します。エクスペクテーションが明確に定義され、追跡可能で、将来の利用に備えられます。
5. データの検証
最後に、定義したエクスペクテーションに対してバッチを検証し、結果を評価します。
# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)検証の実行後、Great Expectationsはデータセットが定義したエクスペクテーションを満たしているかどうかの詳細なレポートを提供します。
{
"success": false,
"results": [
{
"success": true,
"expectation_config": {
"type": "expect_column_values_to_not_be_null",
"kwargs": {
"batch_id": "inventory_parts-inventory_parts_asset",
"column": "inventory_id"
},
"meta": {},
"id": "53d6c42a-d190-412f-a113-783b706531f4"
},
"result": {
"element_count": 580069,
"unexpected_count": 0,
"unexpected_percent": 0.0,
"partial_unexpected_list": [],
"partial_unexpected_counts": [],
"partial_unexpected_index_list": []
},
"meta": {},
"exception_info": {
"raised_exception": false,
"exception_traceback": null,
"exception_message": null
}
},
{
"success": false,
"expectation_config": {
"type": "expect_column_values_to_be_unique",
"kwargs": {
"batch_id": "inventory_parts-inventory_parts_asset",
"column": "part_num"
},
"meta": {},
"id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
},
"result": {
"element_count": 580069,
"unexpected_count": 568352,
"unexpected_percent": 97.98006788847535,
"partial_unexpected_list": [
"48379c01",
"paddle",
"11816pr0005",
"2343",
"3003",
"30176",
"3020",
"3022",
"3023",
"30357",
"3039",
"3062b",
"3068b",
"3069b",
"3069b",
"33291",
"33291",
"3795",
"3941",
"3960"
],
"missing_count": 0,
"missing_percent": 0.0,
"unexpected_percent_total": 97.98006788847535,
"unexpected_percent_nonmissing": 97.98006788847535,
"partial_unexpected_counts": [
{
"value": "3069b",
"count": 2
},
{
"value": "33291",
"count": 2
},
{
"value": "11816pr0005",
"count": 1
},
{
"value": "2343",
"count": 1
},
{
"value": "3003",
"count": 1
},
{
"value": "30176",
"count": 1
},
{
"value": "3020",
"count": 1
},
{
"value": "3022",
"count": 1
},
{
"value": "3023",
"count": 1
},
{
"value": "30357",
"count": 1
},
{
"value": "3039",
"count": 1
},
{
"value": "3062b",
"count": 1
},
{
"value": "3068b",
"count": 1
},
{
"value": "3795",
"count": 1
},
{
"value": "3941",
"count": 1
},
{
"value": "3960",
"count": 1
},
{
"value": "48379c01",
"count": 1
},
{
"value": "paddle",
"count": 1
}
],
"partial_unexpected_index_list": [
0,
3,
4,
5,
6,
7,
8,
9,
10,
11,
12,
13,
14,
15,
16,
17,
18,
19,
20,
21
]
},
"meta": {},
"exception_info": {
"raised_exception": false,
"exception_traceback": null,
"exception_message": null
}
}
],
"suite_name": "inventory_parts_suite",
"suite_parameters": {},
"statistics": {
"evaluated_expectations": 2,
"successful_expectations": 1,
"unsuccessful_expectations": 1,
"success_percent": 50.0
},
"meta": {
"great_expectations_version": "1.2.4",
"batch_spec": {
"batch_data": "PandasDataFrame"
},
"batch_markers": {
"ge_load_time": "20241129T122532.416424Z",
"pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
},
"active_batch_definition": {
"datasource_name": "inventory_parts",
"data_connector_name": "fluent",
"data_asset_name": "inventory_parts_asset",
"batch_identifiers": {
"dataframe": "<DATAFRAME>"
}
}
},
"id": null
}このレポートはデータの品質を詳細に示し、成功と失敗をハイライトします。結果の簡単な説明は次のとおりです。
全体の検証: 検証結果は一部成功でした。エクスペクテーションの50%が合格し、50%が失敗しています。失敗は対処が必要なデータ品質の問題を示します。今回は1つの列が定義したルールを満たしませんでした。
エクスペクテーション1: inventory_idに欠損値がないこと
- 結果: 合格
- 説明:
inventory_id列のすべての値が存在し、NULLや欠損はありません。この列のデータ完全性は良好です。
エクスペクテーション2: part_numが一意であること
- 結果: 失敗
- 説明:
part_num列の97.98%が重複しており、一意な値はごくわずかです。 - ハイライト:
- 重複例には「3069b」や「33291」などがあります。
- これらの重複がどの程度の頻度で出現し、どの行にあるかも表示されるため、問題の特定と修正が容易になります。
もちろん、これはサンプルデータセットであり、合格と不合格の両方の結果が見られるよう、意図的に両方のエクスペクテーションを含めています。
以上です。エンドツーエンドのデータ検証を無事に実行できました。
Great Expectationsをデータパイプラインに統合する
本番環境では、ワークフローに検証を直接組み込み、あらゆる段階でデータ品質を継続的に監視する必要があります。
このセクションでは、Great Expectationsをデータパイプラインに統合する方法について説明します。
以下はイメージを掴むための例であり、ここに含まれない追加の設定が必要になる場合があります。最新の構文は各ツールのドキュメントを参照してください。
ETLツールとの統合
Apache AirflowやPrefectなどの一般的なETLツールとGreat Expectationsを統合するのは比較的容易です。検証ステップをETLプロセスに直接組み込むことで、下流の分析に影響が出る前にリアルタイムでデータの問題を検知して対処できます。
ここでは、PrefectとGreat Expectationsを統合し、自動化されたETLワークフローの一部としてデータ検証を実行する簡単な例を見ていきます。
from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@task
def validate_data():
context = ge.data_context.DataContext()
batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
# Check validation results and raise an alert if validation fails
if not results["success"]:
raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
validation = validate_data()
# Execute the flow
flow.run()この例では、Great Expectationsの検証を実行するタスクを含むPrefectのフローを定義しています。
validate_data()タスクはGreat Expectationsのコンテキストを読み込み、データバッチを取得し、エクスペクテーションスイートを適用します。
データが検証基準を満たさない場合、タスクはアラートを発し、ワークフローを停止して下流のエラーを防ぎます。
継続的なデータ検証
Unix系システムのcronやApache Airflowのようなマネージドサービスなど、さまざまなツールで検証ジョブをスケジュールできます。ここではデータパイプラインのオーケストレーションに適したAirflowを使って、検証の定期実行をスケジュールする方法を示します。
次のようにAirflowのDAG（有向非巡回グラフ）を設定し、Great Expectationsの検証を毎日実行できます。
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
'owner': 'airflow',
'start_date': datetime(2024, 1, 1),
'retries': 1,
}
dag = DAG(
'great_expectations_validation',
default_args=default_args,
schedule_interval='@daily', # Runs once a day
)
# Define the function to run the validation
def run_validation():
context = ge.data_context.DataContext()
batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
return results
# Set up the task in Airflow
validation_task = PythonOperator(
task_id='run_great_expectations_validation',
python_callable=run_validation,
dag=dag,
)
# Set the task in the DAG
validation_taskこの例では、@dailyで1日1回の検証実行をスケジュールするDAGを定義しています。
run_validation()関数は、Great Expectationsのコンテキストを読み込み、定義したエクスペクテーションスイートをデータに対して実行します。
Great Expectationsでのデータ検証のベストプラクティス
スケーラビリティと効率性の観点から、ベストプラクティスに従うことが推奨されます。Great Expectationsによるデータ検証でも同様です。
小さく始めて反復する
まずは基礎的なデータ品質チェックから始め、徐々に拡張していきましょう。最初は基本的なエクスペクテーションに集中する方が、プロセスが複雑になりすぎず、統合やトラブルシューティングがスムーズになります。データセットの理解が深まるにつれて、より複雑な検証を追加できます。
チーム横断で協働する
データ品質は技術的な関心事だけではありません。ビジネスチームと連携してエクスペクテーションを定義し、実装した検証が基礎となるビジネスロジックや目標に合致していることを確認しましょう。部門横断の取り組みにより、データが本来の目的に適い、すべての利害関係者の要件を満たすことが保証されます。
可能な限り自動化する
可能なところは自動化して、データ検証をデータパイプラインに組み込みましょう。自動検証の導入により、手動介入なしでデータ品質を継続的に監視でき、効率が大幅に向上します。
まとめ
お疲れさまでした。Great Expectationsでの設定とデータ検証の方法を学びました。これらの手法は、ワークフローにおける高いデータ品質と透明性の維持に役立ちます。
さらにスキルを伸ばすには、次のリソースをご覧ください。
- ETL and ELT in Python: データの変換と移動を効率的に行う方法を学びます。
- Introduction to Data Quality: データ品質管理の基本を学びます。
- Cleaning Data in Python: 正確性と一貫性を担保するためのデータクリーニング手法を習得します。
- Data Quality Dimensions Cheat Sheet: データ品質ディメンションの便利なリファレンスです。