如何将Airflow DAG配置值传递给GreatExpectationsOperator任务

针对你遇到的Jinja模板未解析、无法将dag_run配置传入GE Checkpoint的问题，以下是三个可行的实操方案：

方案1：用Runtime Batch Request动态构造增量查询 #

直接在Airflow任务中生成包含增量日期的Batch请求，绕过Checkpoint的模板限制，全量测试时用默认日期覆盖所有数据。

from airflow import DAG
from airflow.providers.great_expectations.operators.great_expectations import GreatExpectationsOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1)
}

with DAG('ge_incremental_dq', default_args=default_args, schedule_interval=None) as dag:
    def build_batch_request(**context):
        # 从触发配置取日期，默认全量
        load_start_date = context['dag_run'].conf.get('load_start_date', '1900-01-01')
        return {
            "batch_request": {
                "datasource_name": "your_datasource",
                "data_asset_name": "your_target_table",
                "batch_spec_passthrough": {
                    "query": f"SELECT * FROM your_target_table WHERE created_at >= '{load_start_date}'"
                }
            }
        }

    ge_dq_test = GreatExpectationsOperator(
        task_id='ge_incremental_dq_test',
        expectation_suite_name='your_expectation_suite',
        runtime_batch_request=build_batch_request(),
        data_context_root_dir='/opt/airflow/great_expectations',
        dag=dag
    )

方案2：动态渲染Checkpoint配置文件 #

读取原始Checkpoint配置，用Airflow的dag_run参数替换其中的占位符，再传给GE Operator。

1. 先在checkpoints.yml中预留占位符：

your_checkpoint:
  validations:
    - batch_request:
        datasource_name: "your_datasource"
        data_asset_name: "your_target_table"
        batch_spec_passthrough:
          query: "SELECT * FROM your_target_table WHERE created_at >= '{{LOAD_START_DATE}}'"
      expectation_suite_name: "your_expectation_suite"

2. Airflow DAG中添加渲染逻辑：

import yaml
from airflow import DAG
from airflow.providers.great_expectations.operators.great_expectations import GreatExpectationsOperator
from datetime import datetime
import os

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1)
}

with DAG('ge_incremental_dq', default_args=default_args, schedule_interval=None) as dag:
    def render_checkpoint(**context):
        load_start_date = context['dag_run'].conf.get('load_start_date', '1900-01-01')
        # 读取原始Checkpoint配置
        checkpoint_path = '/opt/airflow/great_expectations/checkpoints/checkpoints.yml'
        with open(checkpoint_path, 'r') as f:
            checkpoint_config = yaml.safe_load(f)
        
        # 替换SQL中的占位符
        checkpoint_config['your_checkpoint']['validations'][0]['batch_request']['batch_spec_passthrough']['query'] = \
            checkpoint_config['your_checkpoint']['validations'][0]['batch_request']['batch_spec_passthrough']['query'].replace('{{LOAD_START_DATE}}', load_start_date)
        
        return checkpoint_config

    ge_dq_test = GreatExpectationsOperator(
        task_id='ge_incremental_dq_test',
        checkpoint_name='your_checkpoint',
        checkpoint_kwargs=render_checkpoint(),
        data_context_root_dir='/opt/airflow/great_expectations',
        dag=dag
    )

方案3：结合GE环境变量模板解析 #

利用GE自身支持的环境变量占位符${VAR_NAME}，先通过Airflow将dag_run参数注入环境变量，再由GE自动解析。

1. 修改checkpoints.yml中的SQL为GE环境变量格式：

your_checkpoint:
  validations:
    - batch_request:
        datasource_name: "your_datasource"
        data_asset_name: "your_target_table"
        batch_spec_passthrough:
          query: "SELECT * FROM your_target_table WHERE created_at >= '${LOAD_START_DATE}'"
      expectation_suite_name: "your_expectation_suite"

2. Airflow DAG中配置环境变量：

from airflow import DAG
from airflow.providers.great_expectations.operators.great_expectations import GreatExpectationsOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1)
}

with DAG('ge_incremental_dq', default_args=default_args, schedule_interval=None) as dag:
    ge_dq_test = GreatExpectationsOperator(
        task_id='ge_incremental_dq_test',
        checkpoint_name='your_checkpoint',
        data_context_root_dir='/opt/airflow/great_expectations',
        # Airflow先渲染Jinja模板，将参数注入环境变量
        env_vars={
            'LOAD_START_DATE': "{{ dag_run.conf.get('load_start_date', '1900-01-01') }}"
        },
        dag=dag
    )

内容的提问来源于stack exchange，提问作者Bazilio