0

我正在尝试将巨大的期望整合到气流管道中。我按照这个网址进行整合。但是我得到错误对象没有属性但是上下文实例具有该属性。这是我的代码。

from pathlib import Path

from airflow import DAG
from airflow.utils.dates import days_ago
from great_expectations_provider.operators.great_expectations import GreatExpectationsOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'data_context_root_dir': Path.cwd() / 'great_expectations/'
}

dag = DAG(dag_id='data-validator',
          default_args=default_args,
          description="Validates data",
          schedule_interval='*/1 * * * *',
          start_date=days_ago(2),
          tags=['data-validator'])

ge_batch_kwargs_pass = GreatExpectationsOperator(
    task_id="ge_batch_kwargs_pass",
    expectation_suite_name="wine_suite",
    batch_kwargs={"path": Path.cwd() / 'airflow/data/output/validation_data.csv', "datasource": "wine_datasource"},
    data_context_root_dir=Path.cwd() / 'great_expectations/',
    dag=dag,
)
ge_batch_kwargs_pass

错误日志

[2021-12-12 14:57:19,452] {great_expectations.py:161} INFO - Ensuring data context exists...
[2021-12-12 14:57:19,453] {great_expectations.py:163} INFO - Data context does not exist, creating now.
[2021-12-12 14:57:19,900] {great_expectations.py:166} INFO - Running validation with Great Expectations...
[2021-12-12 14:57:19,917] {taskinstance.py:1455} ERROR - 'Datasource' object has no attribute 'get_batch'
Traceback (most recent call last):
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1112, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1285, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1315, in _execute_task
    result = task_copy.execute(context=context)
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/great_expectations_provider/operators/great_expectations.py", line 197, in execute
    result = LegacyCheckpoint(
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/great_expectations/checkpoint/checkpoint.py", line 669, in run
    batches_to_validate = self._get_batches_to_validate(self.batches)
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/great_expectations/checkpoint/checkpoint.py", line 731, in _get_batches_to_validate
    batch = self.data_context.get_batch(batch_kwargs, suite)
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/great_expectations/data_context/data_context.py", line 1573, in get_batch
    return self._get_batch_v2(
  File "/Users/rajeshkhadka/PycharmProjects/wine-prediction/venv/lib/python3.9/site-packages/great_expectations/data_context/data_context.py", line 1270, in _get_batch_v2
    batch = datasource.get_batch(
AttributeError: 'Datasource' object has no attribute 'get_batch'
[2021-12-12 14:57:19,923] {taskinstance.py:1496} INFO - Marking task as FAILED. dag_id=data-validator, task_id=ge_batch_kwargs_pass, execution_date=20211210T000000, start_date=20211212T135719, end_date=20211212T135719
[2021-12-12 14:57:20,002] {local_task_job.py:146} INFO - Task exited with return code 1
4

0 回答 0