OI
если сделать свой класс DAG, унаследованный от обычно DAG, то можно просто реплейс импортов аккуратно сделать на всё проекте)
Size: a a a
2021 October 12
EP
Ну как вариант) ну фабрика более очевидный способ)
VK
Ребята, привет. Прошу помощи. Как из Аирфлоу, развернутого на докере, запустить скрипт на SSHOperator
VK
В интернетах не полностью как-то написано, либо я не догоняю
OI
у меня только в failed переходят иногда)
VS
там сейчас, похоже, очереди пулов сидят в общем размере пула
VS
по крайне мере инфа по пулам возвращается именно в таком виде
ДН
Пул сейчас один, но довольно широкий, он не переполняется.
А вот воркеры celery довольно нагружены и скапливаются задачи в очереди сообщений.
Но опять же пока не понятно может ли в этом быть проблема, вроде как таска висит и ты просто ждёшь пока она на какой-нибудь воркер попадёт
А вот воркеры celery довольно нагружены и скапливаются задачи в очереди сообщений.
Но опять же пока не понятно может ли в этом быть проблема, вроде как таска висит и ты просто ждёшь пока она на какой-нибудь воркер попадёт
ДН
Одновременно пара сотен тасок может быть
I
в дефолтовом пуле 128 тасок, то есть это меньше чем пара сотен
Получается, пока даже нет уверенности что таска правда не встаёт очередь, типа может она просто делает это очень медленно? Я бы перепровели всякие parallelism/dag_concurrency/max_active_runs_per_dag и т.п. ещё раз, а так нууу да, лезть в flower и смотреть почему так
Получается, пока даже нет уверенности что таска правда не встаёт очередь, типа может она просто делает это очень медленно? Я бы перепровели всякие parallelism/dag_concurrency/max_active_runs_per_dag и т.п. ещё раз, а так нууу да, лезть в flower и смотреть почему так
VS
С этим решением будет много проблем, особенно в контексте сериализации дагов, ибо когда даг будет создаваться из записи в бд там будет использоваться обычный Dag, он там много где прибит гвоздями. Я через наследование дага хотел как раз pvc создавать, но был послан airflow в пешее эротическое путешествие))
YM
Коллеги, добрый день!
Подскажите, плиз, я вызвал количество записей с xcom
def print_xcom(**kwargs):
ti = kwargs['ti']
val= ti.xcom_pull(task_ids='ss')
return len(val)
Могу я создать луп используя это значение?(Хотел внутри лупа вызвать MssqlOperator, чтобы разделить выгрузку на несколько частей) Может кто то это делал ранее.
Подскажите, плиз, я вызвал количество записей с xcom
def print_xcom(**kwargs):
ti = kwargs['ti']
val= ti.xcom_pull(task_ids='ss')
return len(val)
Могу я создать луп используя это значение?(Хотел внутри лупа вызвать MssqlOperator, чтобы разделить выгрузку на несколько частей) Может кто то это делал ранее.
D
Несколько тасков сделать?
YM
Да
D
dag = DAG('dag',schedule_interval=timedelta(days=1), start_date=days_ago(1))
task_list=list()
for i in range(0, 5):
task_list.append(DummyOperator(task_id=f'task_{i}', dag=dag))
if i:
task_list[i-1] >> task_list[i]YM
Извиняюсь, Я кажется вопрос не правильно задал. Да, я могу ручками указать
for i in range(0, 5):
Но я тут хотел бы автоматизировать, например вместо 5 значение что вернет с xcom
for i in range(0, 5):
Но я тут хотел бы автоматизировать, например вместо 5 значение что вернет с xcom
ME
Хм, тип в разных дагранах разное число тасок?
ME
И оно зависит от первой таски?