Определяя функцию, которая будет использоваться позже как python_callable, почему «ds» включен в качестве первого аргумента функции?Airflow: PythonOperator: зачем включать 'ds' arg?
Например:
def python_func(ds, **kwargs):
pass
Я посмотрел в документации Воздушный поток, но не смогли найти каких-либо объяснений.
Это связано с параметром provide_context=True. Согласно документации на воздушный поток,
Если установлено значение true, Airflow передаст набор аргументов ключевых слов, которые могут использоваться в вашей функции. Этот набор kwargs точно соответствует тому, что вы можете использовать в своих шаблонах jinja. Для этого вам нужно определить ** kwargs в заголовке функции.
ds является одним из этих аргументов ключевых слов и представляет дату исполнения в формате «ГГГГ-ММ-ДД». Для параметров, отмеченных как (шаблонные) в документации, вы можете использовать переменную по умолчанию '{{ ds }}' для передачи даты выполнения. Вы можете прочитать больше о переменных по умолчанию здесь: https://pythonhosted.org/airflow/code.html?highlight=pythonoperator#default-variables
PythonOperator не шаблонные параметры, так что делает что-то вроде python_callable=print_execution_date('{{ ds }}') не будет работать. Для печати даты выполнения внутри отзывной функции вашего PythonOperator, вам придется сделать это как
def print_execution_date(ds, **kwargs): print(ds)
или
def print_execution_date(**kwargs): print(kwargs.get('ds'))
Надеется, что это помогает.
Добро пожаловать в переполнение стека! Это выглядит хорошо продуманным и отформатированным ответом. Спасибо за ваш вклад https://stackoverflow.com/help/how-to-answer – Aron
Если вы определяете свою функцию как 'def something (** kwargs)', тогда вы получаете сообщение об ошибке "неожиданный аргумент ключевого слова 'dag', переданный python_callable (* op_args, ** op_kwargs) ", поэтому кажется, что вам нужно что-то еще, я действительно не понимаю, но используя любой из этих параметров ключевого слова до того, как ** kwargs, похоже, сработает. – Davos