Получение *** Экземпляр задачи не существовал в БД как ошибка при запуске gcs_to_bq в composer

Question

При выполнении следующего скрипта Python с использованием cloud-composer, я получаю *** Task instance did not exist in the DB под задачей gcs2bq. Log in Airflow Code:

import datetime
import os
import csv
import pandas as pd
import pip
from airflow import models
#from airflow.contrib.operators import dataproc_operator
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils import trigger_rule
from airflow.contrib.operators import gcs_to_bq
from airflow.contrib.operators import bigquery_operator

print('''/-------/--------/------/
-------/--------/------/''')
yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())
default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    'start_date': yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    'email_on_failure': False,
    'email_on_retry': False,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': 'data-rubrics'
    #models.Variable.get('gcp_project')
}
try:
  # [START composer_quickstart_schedule]
  with models.DAG(
        'composer_agg_quickstart',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    # [END composer_quickstart_schedule]
      op_start = BashOperator(task_id='Initializing', bash_command='echo Initialized')
      #op_readwrite = PythonOperator(task_id = 'ReadAggWriteFile', python_callable=read_data)
      op_load = gcs_to_bq.GoogleCloudStorageToBigQueryOperator( \
task_id='gcs2bq',\
bucket='dr-mockup-data',\
source_objects=['sample.csv'],\
destination_project_dataset_table='data-rubrics.sample_bqtable',\
schema_fields = [{'name':'a', 'type':'STRING', 'mode':'NULLABLE'},{'name':'b', 'type':'FLOAT', 'mode':'NULLABLE'}],\
write_disposition='WRITE_TRUNCATE',\
dag=dag)
      #op_write = PythonOperator(task_id = 'AggregateAndWriteFile', python_callable=write_data)
      op_start >> op_load

Answer 1

Хорошо, это был глупый вопрос с моей стороны, и я прошу прощения у всех, кто потратил здесь время.У меня был Dag, из-за которого тот, с которого я стрелял, всегда был в очереди.Кроме того, я не написал правильное значение в destination_project_dataset_table.Спасибо и извинения всем, кто провел время.

Answer 2

UPDATE

Можете ли вы удалить dag=dag из задачи gcs2bq, поскольку вы уже используете with models.DAG, и снова запустить dag?

---

Это может быть потому, что у вас есть динамическая дата начала. Ваш start_date никогда не должен быть динамичным. Прочитайте этот FAQ: https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date

Мы не рекомендуем использовать динамические значения в качестве start_date, особенно datetime.now (), так как это может привести к путанице. Задание запускается после закрытия периода, и теоретически DAG @hourly никогда не достигнет часа спустя, поскольку now () движется вперед.

Сделайте ваш start_date статичным или используйте утилиты / макросы Airflow:

import airflow
args = {
    'owner': 'airflow',
    'start_date': airflow.utils.dates.days_ago(2),
}