来自另一个文件的气流导入任务依赖项答案

【问题标题】：Airflow import task dependency from another file来自另一个文件的气流导入任务依赖项
【发布时间】：2021-02-15 17:28:47
【问题描述】：

我的 Airflow 有 4 个任务。 t1, t2, t3, t4 任务 id 为 task_id1, task_id2, task_id3, task_id4

但是，任务的执行顺序：t1 >> t2 >> t3 >> t4 将从文本文件中读取。

Example: the text file will have:
t1
t2
t3
t4

or

t1
t2
t4
t3

So the order will be: t1 >> t2 >> t3 >> t4   or  t1 >> t2 >> t4 >> t3

But if i try:
f= open("file_name.txt")
lines = f.readlines()
task1 = lines[0].rstrip()
task2 = lines[1].rstrip()
task3 = lines[2].rstrip()
task4 = lines[3].rstrip()

and then use:
task1 >> task2 >> task3 >> task4

I get error:
×Broken DAG: unsupported operand type(s) for >>: 'str' and 'str

有什么建议吗...？

实际代码：

from datetime import timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.helpers import chain
from airflow.utils.dates import days_ago
from airflow import AirflowException
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': days_ago(2),
}

# Create DAG instance
dag = DAG(
'test',
default_args=default_args,
description='A simple tutorial DAG',
catchup=False,
schedule_interval=timedelta(seconds=20),
)


# First task
t1 = BashOperator(
task_id='task1',
bash_command='echo "task 1"',
dag=dag,
)

# Second task
t2 = BashOperator(
task_id='task2',
bash_command='echo "task2"',
dag=dag,
)

# same for t3 and t4

f= open("file_name.txt")
lines = f.readlines()
task1 = lines[0].rstrip()
task2 = lines[1].rstrip()
task3 = lines[2].rstrip()
task4 = lines[3].rstrip()
f.close()

task1 >> task2 >> task4 >> task3

file_name.txt 可以有： t1 t2 t4 t3

我想要这个订单，但这个订单必须来自外部文件而不是同一个文件。

【问题讨论】：

操作符之间设置依赖关系。运算符是 python 类。目前尚不清楚您在文本文件中到底有什么。它是您的 dag 代码还是只是您应该阅读并应用于您的代码的依赖项列表？
可能，首先你需要了解这个操作符是做什么的（wiki.python.org/moin/BitwiseOperators）。
@RomanZhuravlev 这不是 python 按位。这是气流位移airflow.apache.org/docs/apache-airflow/stable/…
啊，好吧，我的错。但是，OP的数据显然是str类型
文本文件包含我想要阅读并应用于代码的依赖项列表。如果文本文件在第 1 行有 t1，在第 2 行有 t2，则气流必须工作：t1 >> t2 如果文本文件在第 1 行有 t2，在第 2 行有 t1，则气流必须工作：t2 >> t1

标签： python dependencies airflow scheduler directed-acyclic-graphs

【解决方案1】：

This thing helped:
def str_to_class(classname):
    return getattr(sys.modules[__name__], classname)

task1 = ''
task2 = ''
task3 = ''
task4 = ''

f= open("(path_to_scheduler_list.txt")
lines = f.readlines()
task1 = lines[0].rstrip()
task2 = lines[1].rstrip()
task3 = lines[2].rstrip()
task4 = lines[3].rstrip()
f.close()
task1 = str_to_class(task1)
task2 = str_to_class(task2)
task3 = str_to_class(task3)
task4 = str_to_class(task4)

task1 >> task2 >> task3 >> task4

【讨论】：