в образа с логикой

Надо переварить сказанное. Docker для каждого дага? Я правильно понял?)

докер для каждой таски в даге, если у вас все очень стремительно и сушить мозг некогда - можно делать даг из одной таски и тогда в вашем случае это равносильно "докер для каждого дага" =)

В общем готовить много оперативки ) норм )

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

Вообще никакой проблемы в том, чтобы его пересобирать раз в день 🤷🏻‍♂️
Сделайте CI/CD себе, пусть раннер собирает и кладёт в реестр

Это норма, удобно

Для меня смысл эйрфлоу в т.ч. чтобы таски были.
В общем мне подходит концепция - внутри Дага несколько тасков-докеров

данный подход на самом деле очень удобен, мы например сейчас образ с airflow обновляем только когда надо обновить сам airflow или его провайдеры

а все остальное пакуем в образа которые параметризуются входными параметрами, ну и все собирается в gitlab

недавно даже про это рассказывали на смартдате

=)

Спасибо

коллеги, не поможете с простым вопросом, пишу в базу, подключаюсь
к ней  dest = PostgresHook(postgres_conn_id ='pg_write')

приходит ошибка
 File "/home/airflow/.local/lib/python3.6/site-packages/psycopg2/__init__.py", line 122, in connect
   conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
psycopg2.OperationalError: FATAL:  database "datamodel" does not exist

база datamodel - есть
в чем может быть косяк?

А как задаётся коннекшн, через ui?

да, через UI, стандартно.
задача достаточно простая из одной postgres базы прочитать и в другую написать, и что-то подвис на ней.

А если задать путь как jdbc урл через окружение (AIRFLOW_CONN_PG_WRITE='psycopg2+psql://blablabra'), то будет работать?

Я помню там лично для меня была какая-то странность в названиях полей в ui, чё-то типа что базу надо было в схему написать что ли, я бы попробовал туда смотреть

попробую, а это где писать? в airflow.cfg?
вроде написал самое простое
import datetime

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.hooks.postgres_hook import PostgresHook


with DAG(
   dag_id="postgres_dag",
   start_date=datetime.datetime(2021, 10, 15),
   schedule_interval="@daily",
   catchup=False,
) as dag:
      src = PostgresHook(postgres_conn_id ='Read_pg')
      dest = PostgresHook(postgres_conn_id ='pg_write')
      src_conn = src.get_conn()
      cursor = src_conn.cursor()
      cursor.execute("select * from table_read")
      dest_conn = dest.get_conn()
      dest_cursor = dest_conn.cursor()
      dest.insert_rows(table="staging.table_write", rows=cursor)

Мммм ну вот так вот точно не надо делать, получается что код работы с хуками будет исполняться при каждом чтении даг файла. Эта логика должна быть внутри питон оператора, внутри дага только структура

Писать надо в переменных окружения, это не airflow.cfg, это в том контексте где запускается щедудер и вебсервер

Если в докере, например, то

ENV AIRFLOW_CONN_PG_WRITE='...'

Если в шелл то перед запуском команд
export AIRFLOW_CONN_PG_WRITE='...'