тоже не пойму, но я нашел, сейчас смотрю код

?

без понятия

я помню, что в 1.10 там точно были проблемы, которые не смогли мы решить своими силами, но вроде во второй версии должны были поправить, но это не точно

так, нашел в логах ERROR - [Errno 2] No such file or directory: 'yarn': 'yarn'
видимо не копируются переменные окружения, но под этим пользователем в консоли все находится.
Вопрос как передать их автоматом, как PATH который прописан в тот же ~/.profile передать в процесс

if self._yarn_application_id:
               kill_cmd = f"yarn application -kill {self._yarn_application_id}".split()
               env = None #<— вот это зачем в коде???
               if self._keytab is not None and self._principal is not None:
                   # we are ignoring renewal failures from renew_from_kt
                   # here as the failure could just be due to a non-renewable ticket,
                   # we still attempt to kill the yarn application
                   renew_from_kt(self._principal, self._keytab, exit_on_fail=False)
                   env = os.environ.copy()
                   env["KRB5CCNAME"] = airflow_conf.get('kerberos', 'ccache')

               yarn_kill = subprocess.Popen(
                   kill_cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE
               )

Коллеги, всем привет. Подскажите, пожалуйста, хочу воспользоваться on_failure_callback. Прописал вызов в default_args

Что хочу получить:
context['task_instance'].dag_id,  context['task_instance'].task_id,
context['exception']

Но в итоге в task_id пусто. Мне нужно для каждой задачи прописывать on_failure_callback? Я думал, что default_args передаёт параметры на все таски

Вопрос странно поставлен. Если вы говорите, что task_id пустой значит ваш коллбек отработал и вы как-то увидели, что он пустой, так? А если он отработал, значит он передался как параметр в задачу.

Только что разобрался. Не указал в format передачу второго параметра, который отвечал за task_id)
Надо бы проснуться))
Спасибо

Главное, что быстро :)

всем привет,
я вчера мучался с запуском spark job на yarn cluster, было две проблемы:
1. в случае если ярн убивает задачу, то в airflow она продолжала висеть как запущенная
2. если я останавливаю задачу в airflow то в ярн она продолжает работать

первую проблема была из-за того, что если master указан как yarn (рекомендованный метод запуска)  а не через spark://master, то не отслеживалось состояния драйвера
вторая что не передавалось env в запуск yarn (я вчера кусок кода присылал), это я тоже пофиксил строкой в исходниках+передавая PATH из  конфига airflow (path = Variable.get("PATH")),  но вот такая ручная передача выглядит кривовато.
Я новичок в airflow, а как вообще принято передавать вот такие системные переменные окружения?

Всем привет, а кто то использует airflow для ml? Вопрос такой, как используете или если не используете то почему.

airflow запускает sshoperator, внутри которого вызов либы, решающей любую etl-задачу, в том числе скоринг мл-моделью или её автоматическое переобучение

от такого варианта ушли, хочется без bash/ssh

Без ssh нужен кубер/селери, иначе например вам нужно мл обучить на гпу, а у вас только одна нода с гпу, и эйрфлоу не на ней, и начинаются костыли

пнл

тогда вернемся к bash/ к куберу пока не готовы

У нас одна либа на всё, её эйрфлоу только вызывает и прокидывает конфиг

Основной плюс - можно отлаживать етл и скоринг без эйрфлоу