Ошибка воздушного потока - получен неожиданный аргумент ключевого слова 'min' - PullRequest
Новая
       81

Ошибка воздушного потока - получен неожиданный аргумент ключевого слова 'min'

0 голосов
/ 19 марта 2020

Я пытаюсь запустить очень простой тестовый DAG, чтобы получить представление о базовых c функциональных возможностях GCP Cloud Composer, но каждый раз, когда я запускаю DAG, появляется неприятная ошибка, и я не могу показаться чтобы найти любую информацию о том, как ее решить.

Ошибка: 

2020-03-18 22:20:56,627] {taskinstance.py:1059} ERROR - __init__() got an unexpected keyword argument 'min'@-@{"workflow": "notebook-test", "task-id": "notebook-test", "execution-date": "2020-03-18T22:20:41.232043+00:00"}
Traceback (most recent call last):
  File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 930, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 113, in execute
    return_value = self.execute_callable()
  File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 118, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/home/airflow/gcs/dags/test.py", line 44, in execute_nb
    parameters=params
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/execute.py", line 104, in execute_notebook
    **engine_kwargs
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/engines.py", line 49, in execute_notebook_with_engine
    return self.get_engine(engine_name).execute_notebook(nb, kernel_name, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/engines.py", line 341, in execute_notebook
    nb_man.notebook_start()
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/engines.py", line 69, in wrapper
    return func(self, *args, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/engines.py", line 198, in notebook_start
    self.save()
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/engines.py", line 69, in wrapper
    return func(self, *args, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/engines.py", line 139, in save
    write_ipynb(self.nb, self.output_path)
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/iorw.py", line 397, in write_ipynb
    papermill_io.write(nbformat.writes(nb), path)
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/iorw.py", line 128, in write
    return self.get_handler(path).write(buf, path)
  File "/opt/python3.6/lib/python3.6/site-packages/papermill/iorw.py", line 316, in write
    multiplier=self.RETRY_MULTIPLIER, min=self.RETRY_DELAY, max=self.RETRY_MAX_DELAY
TypeError: __init__() got an unexpected keyword argument 'min'

, и мой код DAG: 

import airflow
import papermill as pm
from datetime import timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator


default_args = {
    'owner': 'airflow',
    'start_date': airflow.utils.dates.days_ago(1),
    'end_date': None,
    'retries': 0,    
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    dag_id="notebook-test",
    description="a test",
    default_args=default_args,
    catchup=True,
    schedule_interval=None,
    dagrun_timeout=(timedelta(seconds=30))
)

NB_PATH = "gs://BUCKET/data/"

params = {}


def execute_nb():
    input_nb = NB_PATH + "test.ipynb"
    output_nb = NB_PATH + "test_ran.ipynb"

    pm.execute_notebook(
        input_nb,
        output_nb,
        parameters=params
    )


op = PythonOperator(
    task_id="notebook-test",
    python_callable=execute_nb,
    dag=dag
)

op

Одно решение, которое я уже пробовал https://github.com/nteract/papermill/issues/445 должен был обновить версию Tenacity, но добавление этого на вкладку PyPi Packages моей среды Cloud Composer ничего не решило.

Любая помощь будет оценена, спасибо!

РЕДАКТИРОВАТЬ: Версия изображения composer -1.9.2-airflow.1.10.6.

1 Ответ

1 голос
/ 20 марта 2020

Получается, что проблема связана с указанным путем.

Мне пришлось добавить import os, а также from pathlib import Path, а затем сделать мою переменную

NB_PATH = str(Path(os.path.abspath(__file__)).parents[1]) + "/data".

Это также потребовало от меня добавить jupyter как зависимость PyPi для бумажной фабрики работает правильно, но, похоже, она работает сейчас!

...