> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Orchestrate Great Expectations with Airflow

[Great Expectations](https://greatexpectations.io) (GX) is an open source Python-based data validation framework. You can test your data by expressing what you "expect" from it as simple declarative statements in JSON or YAML, then run validations using those [Expectation Suites](https://docs.greatexpectations.io/docs/core/define_expectations/organize_expectation_suites) against data [SQL data](https://docs.greatexpectations.io/docs/core/connect_to_data/sql_data), [Filesystem Data](https://docs.greatexpectations.io/docs/core/connect_to_data/filesystem_data) or a [pandas DataFrame](https://docs.greatexpectations.io/docs/core/connect_to_data/dataframes). The [airflow-provider-great-expectations](https://great-expectations.github.io/airflow-provider-great-expectations/latest/getting-started/) package provides operators for running Great Expectations validations directly in your Dags.

<Warning>
  Version 1.0.0 of the provider introduces three specialized operators that replace the legacy `GreatExpectationsOperator`:
</Warning>

| Operator                       | Use case                                                             |
| ------------------------------ | -------------------------------------------------------------------- |
| `GXValidateDataFrameOperator`  | Validate in-memory Spark or Pandas DataFrames                        |
| `GXValidateBatchOperator`      | Validate data not in memory using a BatchDefinition                  |
| `GXValidateCheckpointOperator` | Most feature-rich: supports triggering actions on validation results |

**Requirements**: Python 3.10+, Great Expectations 1.7.0+, Apache Airflow 2.1+.

Install with:

```text wrap theme={null}
pip install airflow-provider-great-expectations
```

Each operator has its own import path:

```python wrap theme={null}
from great_expectations_provider.operators.validate_dataframe import GXValidateDataFrameOperator
from great_expectations_provider.operators.validate_batch import GXValidateBatchOperator
from great_expectations_provider.operators.validate_checkpoint import GXValidateCheckpointOperator
```

#### Choose the right operator

When deciding which operator fits your use case, consider:

1. **Where is your data?** In memory as a DataFrame, or in an external data source?
2. **Do you need to trigger actions?** Such as sending notifications or updating external systems based on validation results.
3. **What Data Context do you need?** Ephemeral for stateless validations, or persistent to track results over time.

| Scenario                                                      | Recommended operator           |
| ------------------------------------------------------------- | ------------------------------ |
| Data already in memory as Pandas or Spark DataFrame           | `GXValidateDataFrameOperator`  |
| Data in a database, warehouse, or file system                 | `GXValidateBatchOperator`      |
| Need to trigger Slack notifications, emails, or other actions | `GXValidateCheckpointOperator` |
| Want full GX Core features with ValidationDefinitions         | `GXValidateCheckpointOperator` |

#### `GXValidateDataFrameOperator`

Use this operator when your data is already in memory as a Pandas or Spark DataFrame. This is the simplest option, you only need a DataFrame and your expectations.

<Accordion title="GXValidateDataFrameOperator example">
  ```python expandable wrap theme={null}
  from great_expectations_provider.operators.validate_dataframe import GXValidateDataFrameOperator

  def configure_dataframe():
      """Load data into a DataFrame using an Airflow hook."""
      from airflow.providers.common.sql.hooks.sql import DbApiHook

      hook = DbApiHook.get_hook("my_db_conn")
      return hook.get_pandas_df("SELECT * FROM daily_planet_report")

  def configure_expectations(dataframe):
      """Define expectations for the DataFrame."""
      import great_expectations as gx

      suite = gx.ExpectationSuite(name="daily_report_suite")
      suite.add_expectation(
          gx.expectations.ExpectColumnValuesToNotBeNull(column="planet_name")
      )
      suite.add_expectation(
          gx.expectations.ExpectColumnValuesToNotBeNull(column="total_passengers")
      )
      suite.add_expectation(
          gx.expectations.ExpectColumnValuesToBeBetween(
              column="total_net_fare_usd", min_value=0
          )
      )
      suite.add_expectation(
          gx.expectations.ExpectColumnValuesToBeBetween(
              column="total_discounts_usd", min_value=0
          )
      )
      return suite

  _validate_dataframe = GXValidateDataFrameOperator(
      task_id="validate_with_gx_dataframe",
      configure_dataframe=configure_dataframe,
      configure_expectations=configure_expectations,
      result_format="SUMMARY",
      context_type="ephemeral",
  )
  ```
</Accordion>

<Info>
  **configure\_expectations signature**

  The `configure_expectations` callable receives the DataFrame as its first argument and must return an `Expectation` or `ExpectationSuite`.
</Info>

#### `GXValidateBatchOperator`

Use this operator when your data isn't in memory. You configure GX to connect directly to your data source by defining a `BatchDefinition`. This approach works with databases, warehouses, and file systems.

<Tip>
  The provider includes helper functions to build connection strings from Airflow connections, so you don't need to duplicate credentials.
</Tip>

<Accordion title="GXValidateBatchOperator example (SQL)">
  ```python expandable wrap theme={null}
  from great_expectations_provider.operators.validate_batch import GXValidateBatchOperator
  from great_expectations_provider.common.external_connections import build_snowflake_connection_string

  SNOWFLAKE_CONN_ID = "snowflake_default"


  def configure_batch_definition(context):
      """Configure a batch definition for a SQL table."""
      connection_string = build_snowflake_connection_string(
          conn_id=SNOWFLAKE_CONN_ID,
          schema="my_schema"
      )

      data_source = context.data_sources.add_sql(
          name="snowflake_ds",
          connection_string=connection_string,
      )

      table_asset = data_source.add_table_asset(
          name="daily_report_table",
          table_name="daily_planet_report",
      )

      return table_asset.add_batch_definition_whole_table(name="full_table")


  def configure_expectations(context):
      """Define expectations and add them to the context."""
      import great_expectations.expectations as gxe
      from great_expectations import ExpectationSuite

      return context.suites.add_or_update(
          ExpectationSuite(
              name="daily_report_batch_suite",
              expectations=[
                  gxe.ExpectColumnValuesToNotBeNull(column="planet_name"),
                  gxe.ExpectColumnValuesToNotBeNull(column="total_passengers"),
                  gxe.ExpectColumnValuesToBeBetween(
                      column="total_net_fare_usd", min_value=0
                  ),
                  gxe.ExpectColumnValuesToBeBetween(
                      column="total_discounts_usd", min_value=0
                  ),
              ],
          )
      )


  _validate_batch = GXValidateBatchOperator(
      task_id="validate_with_gx_batch",
      configure_batch_definition=configure_batch_definition,
      configure_expectations=configure_expectations,
      result_format="SUMMARY",
      context_type="ephemeral",
  )
  ```
</Accordion>

<Accordion title="GXValidateBatchOperator example (file-based)">
  For file-based data (CSV, Parquet), use `add_pandas_filesystem` instead:

  ```python wrap theme={null}
  def configure_batch_definition(context):
      """Configure a batch definition to read from a CSV file."""
      from pathlib import Path

      data_source = context.data_sources.add_pandas_filesystem(
          name="my_datasource",
          base_directory=Path("/path/to/data"),
      )
      csv_asset = data_source.add_csv_asset(name="daily_report_csv")
      return csv_asset.add_batch_definition_path(
          name="daily_report_batch",
          path="daily_report.csv",
      )
  ```
</Accordion>

<Info>
  **configure\_expectations signature**

  For `GXValidateBatchOperator`, the `configure_expectations` callable receives the GX `context` as its first argument (not the DataFrame). Use `context.suites.add_or_update()` to register your suite.
</Info>

#### `GXValidateCheckpointOperator`

Use this operator when you need the full power of GX Core, including the ability to trigger actions based on validation results. This requires the most configuration. You define a `Checkpoint` with a `BatchDefinition`, `ExpectationSuite`, and `ValidationDefinition`.

<Accordion title="GXValidateCheckpointOperator example">
  ```python expandable wrap theme={null}
  from great_expectations_provider.operators.validate_checkpoint import GXValidateCheckpointOperator
  from great_expectations_provider.common.external_connections import build_snowflake_connection_string

  SNOWFLAKE_CONN_ID = "snowflake_default"

  def configure_checkpoint(context):
      """Configure a full GX checkpoint with validation and actions."""
      import great_expectations.expectations as gxe
      from great_expectations import Checkpoint, ExpectationSuite, ValidationDefinition

      connection_string = build_snowflake_connection_string(
          conn_id=SNOWFLAKE_CONN_ID,
          schema="my_schema"
      )

      batch_definition = (
          context.data_sources.add_sql(
              name="snowflake_ds",
              connection_string=connection_string,
          )
          .add_table_asset(
              name="daily_report_table",
              table_name="daily_planet_report",
          )
          .add_batch_definition_whole_table(name="full_table")
      )

      expectation_suite = context.suites.add(
          ExpectationSuite(
              name="daily_report_checkpoint_suite",
              expectations=[
                  gxe.ExpectColumnValuesToNotBeNull(column="planet_name"),
                  gxe.ExpectColumnValuesToNotBeNull(column="total_passengers"),
                  gxe.ExpectColumnValuesToBeBetween(
                      column="total_net_fare_usd", min_value=0
                  ),
                  gxe.ExpectColumnValuesToBeBetween(
                      column="total_discounts_usd", min_value=0
                  ),
              ],
          )
      )

      validation_definition = context.validation_definitions.add(
          ValidationDefinition(
              name="daily_report_validation",
              data=batch_definition,
              suite=expectation_suite,
          )
      )

      return context.checkpoints.add(
          Checkpoint(
              name="daily_report_checkpoint",
              validation_definitions=[validation_definition],
              actions=[],  # Add SlackNotificationAction, EmailAction, etc.
          )
      )

  _validate_checkpoint = GXValidateCheckpointOperator(
      task_id="validate_with_gx_checkpoint",
      configure_checkpoint=configure_checkpoint,
      context_type="ephemeral",
  )
  ```
</Accordion>