Невозможно отладить ошибку воздушного потока. Попытка использовать воздушный поток для построения конвейера данных на GCP - PullRequest
Новая
       7

Невозможно отладить ошибку воздушного потока. Попытка использовать воздушный поток для построения конвейера данных на GCP

0 голосов
/ 01 декабря 2018

Я пытаюсь определить, что может быть причиной указанной ниже проблемы (Воздушный поток)?

В основном я написал тестовую группу DAG, и его главная задача - прочитать данные из BigQuery и записатьэто в новую таблицу. Я пытался найти об этом, но я не могу выяснить, что может быть причиной этого.Я даже не уверен, правильно ли работает мой gcp_connection.Я не знаю, как это проверить.Любая помощь с благодарностью! 

Traceback (most recent call last):
  File "/anaconda3/lib/python3.6/site-packages/airflow/models.py", line 1659, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/anaconda3/lib/python3.6/site-packages/airflow/operators/subdag_operator.py", line 103, in execute
    executor=self.executor)
  File "/anaconda3/lib/python3.6/site-packages/airflow/models.py", line 4214, in run
    job.run()
  File "/anaconda3/lib/python3.6/site-packages/airflow/jobs.py", line 203, in run
    self._execute()
  File "/anaconda3/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
    return func(*args, **kwargs)
  File "/anaconda3/lib/python3.6/site-packages/airflow/jobs.py", line 2547, in _execute
    raise AirflowException(err)
airflow.exceptions.AirflowException: ---------------------------------------------------
Some task instances failed:
{('test_oscope.test_oscope', 'create_if_not_exists', datetime.datetime(2016, 6, 1, 0, 0, tzinfo=<Timezone [UTC]>), 1), ('test_oscope.test_oscope', 'fill', datetime.datetime(2016, 6, 1, 0, 0, tzinfo=<Timezone [UTC]>), 1)}

1 Ответ

0 голосов
/ 02 декабря 2018

Крючки в Airflow очень полезны для использования в интерактивных средах, таких как iPython или Jupyter Notebook.

Например: 

from airflow.contrib.hooks.gcs_hook import GoogleCloudStorageHook

GCSHook = GoogleCloudStorageHook(google_cloud_storage_conn_id='google_conn_id')
GCSHook.get_conn() # This will check if your GCP connection is working correctly or not
...