airflow_dag_great_expectations_checkpoint_validation.py

1import os
2from pathlib import Path
3from datetime import datetime
4
5from airflow import DAG
6from great_expectations_provider.operators.great_expectations import GreatExpectationsOperator
7
8# This path should point to your Great Expectations directory
9base_path = Path(__file__).parents[2]
10ge_root_dir = os.path.join(base_path, "great_expectations")
11
12with DAG(
13    dag_id="example_great_expectations_dag",
14    start_date=datetime(2021, 1, 1),
15    catchup=False,
16    schedule_interval=None,
17) as dag:
18
19    # Task to run a Great Expectations checkpoint
20    ge_checkpoint_pass = GreatExpectationsOperator(
21        task_id="ge_checkpoint_pass",
22        data_context_root_dir=ge_root_dir,
23        checkpoint_name="taxi.pass",  # Replace with your checkpoint name
24        return_json_dict=True,
25    )
26
27    # Example of a task that is expected to fail validation
28    ge_checkpoint_fail = GreatExpectationsOperator(
29        task_id="ge_checkpoint_fail",
30        data_context_root_dir=ge_root_dir,
31        checkpoint_name="taxi.fail",  # Replace with your checkpoint name
32        fail_task_on_validation_failure=True,
33    )
34
35    ge_checkpoint_pass >> ge_checkpoint_fail