R
Size: a a a
2021 December 13
J
поетри?
АБ
Всем привет вопрос насчет AIRFLOW, есть следующая задача, запускать определенную цепочку тасков, для каждого сообщения из PubSub'a, то, есть dag запускается, берет какой-то батч, условно 10 сообщений, и создается 10 цепочек тасков для каждого сообщения. В данный момент есть следующая проблема, что во время создания Dag'a airflow не знает о существование сообщений, соответсвено цепочка не выстраивается, можете подсказать идею в какую сторону искать?
RQ
а в каком моменте вы тяните сообщения из pubsub?
RQ
попадая во внутрь контекста
with Dag() as dag:у вас уже есть список/лист этих сообщений?
АБ
да
RQ
with DAG(
dag_id,
schedule_interval='0 7 * * *',
start_date=datetime(2021, 4, 20, 0, 0),
catchup=False,
) as dag:
messages = get_messages_from_pub_sub() # list
messages_task = {}
for message in messages:
messages_task[message['id']] = (PythonOperator(
task_id=f'message_task_{message['id']}',
provide_context=False,
python_callable=task_handler,
op_kwargs={'param1': message['param1'], 'param2': message['param2']},
))
RQ
messages = get_messages_from_pub_sub() # list
ваши данные из pubsub
АБ
то, есть разница где я вызываю функцию get_messages_from_pub_sub?
RQ
далее
for message in messages:
task_start >> messages_task[message['id']] >> task_end
RQ
я пока не совсем понимаю весь контекст вашей задачи, попробуйте пока так
RQ
В чате есть Динсултан и Зульфикар, они шарят в Airflow
RQ
У вас должен вырисоваться такой граф, я его делал для Continuous Delivery ML-моделей
Вместо моих городов, у вас должны отрисоваться messages
Вместо моих городов, у вас должны отрисоваться messages
АБ
Да, правильно поняли
RQ
Гуд. Потом отпишите как решили. Интересно