Внутри папки airflow, вручную создайте папку dags, или в airflow.cfg можно указать любой другой путь к дагам

На самом деле достаточно даже просто airflow запустить без параметров и конфиг будет создан

Спасибо! Я так и подумал что пути скорее всего находятся в конфигурации где то

Есть PythonVirtualOperator, внутри него могут существовать зависимости из requirements.txt. Как это все работает точно не скажу, но недостающие пакеты, как я понял, прямо налету устанавливаются пр необходимости pip-ом. Вот репозиторий для пипа где задаётся я не знаю...

магия какая то

Но это работает, сам использую))

Так что в разных тасках в одном даге можно подружить код с разными зависимостями

Единственное я ещё не понял где логи хранятся при этом, потому что у виртуального оператора они только после успешной отработки таски выскакивают..

Ребят всем привет. А у кого airflow в кубере вертится, вы поды удаляете сразу после выполнения, или потом все завершенные подчищаете (например, отдельным дагом)?

сразу, на проде
на стейдже можно не удалять чисто для дебага

Понял, спасибо. Просто столкнулся с такой проблемой, что если по какой-то причине под не стартует и сразу удаляется, то увидеть в чем причина не получится. А если его не удалять при ошибке, то при ретрае он не запустится, т.к под с таким именем уже существует. Выглядит логично чтобы все нормально оттестить на стейдже, но всякое же бывает )

Привет!

У меня есть два вопроса:
1. Как вы передаете параметры в sql?
2. Тут про сабдаги пишут, что нужно держаться от них подальше. Это до сих пор актуально? Как вы реализуете трехступенчатую вложенность?

1. в большинстве sql-операторов работает jinja-темплейтинг, туда можно передавать всякие даты, xcom и прочее

Добро пожаловать в самое дружелюбное комьюнити.

ребята, кто то сталкивался с подобной ошибкой когда забрасываешь DAG? Broken DAG: [/root/airflow/dags/spark_dag.py] Traceback (most recent call last):
 File "/usr/local/lib/python3.8/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py", line 32, in <module>
   from airflow.kubernetes import kube_client
 File "/usr/local/lib/python3.8/site-packages/airflow/kubernetes/kube_client.py", line 101, in <module>
   ) -> client.CoreV1Api:
NameError: name 'client' is not defined

airflow развернут на хост машине, без kubernetes

В версиях 1.* (не знаю, как в 2) дефолтный экзекьютор для сабдагоператоров – SequentialExecutor, по идее executor можно передать параметром при объявлении оператора, но они настойчиво не советуют это менять.
Короче, сабдаги не полетят выполняться на celery-воркерах

https://issues.apache.org/jira/browse/AIRFLOW-74

хмм, у меня летают

С заданным CeleryExecutor-ом? Sequential разрешает только одному таску быть запущенному в моменте

ага