Xcom pull возвращает NameError на `ti`

Я пытаюсь вставить в xcom значение с ключом last_date в last_date_task, а затем вытащить его из xcom во второй задаче ga_wh_task. когда я тестирую этот даг, запустив python dag.py, он возвращает ошибку в этой строке, например:

    provide_context=True, since={{ti.xcom_pull(task_ids="last_date_task", key='last_date')}})
NameError: name 'ti' is not defined

Что я здесь делаю не так? Я пробовал использовать task_instance вместо ti и context['ti'] вместо kwargs['ti']

Вот dag-файл:

default_args = {
    'owner': 'me',
    'start_date': dt.datetime(2017, 10, 30),
    'retries': 1,
    'retry_delay': dt.timedelta(minutes=10),
    'provide_context' : True,
}



def get_last_date(**kwargs):
    kwargs['ti'].xcom_push(key='last_date', value='2018-11-15')
    return True



with DAG('ga_mysql_dag2',
         default_args=default_args,
         schedule_interval=None,
         catchup=False,
         ) as dag:

    last_date_task = PythonOperator(task_id='last_date_task', python_callable=get_last_date, provide_context=True)

    ga_wh_task = GoogleAnalyticsReportingToMySqlOperator(task_id='ga_wh_task', google_analytics_conn_id='google_analytics', key_file=key_file,\
                                        view_id=view_id, until=until, dimensions=dimensions, metrics=metrics, database=database,\
                                        table = table, mysql_conn_id = mysql_conn_id,
                                        provide_context=True, since={{ti.xcom_pull(task_ids="last_date_task", key='last_date')}})
    sleep = BashOperator(task_id='sleep', bash_command='sleep 10')

# Dependencies
last_date_task >> ga_warehouse_task >> sleep

python-3.x airflow airflow-scheduler apache-airflow-xcom
person Nihal Sangeeth    schedule 16.11.2018    source источник

Ответы (1)


arrow_upward
2
arrow_downward

Похоже, что GoogleAnalyticsReportingToMySqlOperator - это созданный вами оператор.

since параметр должен быть строкой. Так что измените его на since="{{ti.xcom_pull(task_ids='last_date_task', key='last_date')}}"

person kaxil    schedule 16.11.2018
comment
Пробовал. Теперь у меня есть NameError на _1 _...._ 2_ - person Nihal Sangeeth; 16.11.2018
comment
GoogleAnalyticsReportingToMySqlOperator - это оператор, который я создал, да. - person Nihal Sangeeth; 16.11.2018
comment
Попробуйте с since="{{ti.xcom_pull(task_ids='last_date_task', key=None)}}". Это должно работать - person kaxil; 16.11.2018