在气流中运行时的自定义操作员 XCom

Question

我知道PythonOperator/BashOperator我们可以使用Xcom进行交流。

例如

def func(**context):
context['task_instance'].xcom_pull()

但是，我想知道如何在运行时访问自定义运算符的xcom。

我的运算符如下所示：

class ECHOXOperator(BaseOperator):
@apply_defaults
def __init__(self, x, *args, **kwargs):
self.x = x
super(ECHOXOperator, self).__init__(*args, **kwargs)
def execute(self, context):
print(self.x)

所以在我的DAG中：

我能做到

task2 = ECHOXOperator(x = 'Hello")

而且效果很好。但是如何从上游任务访问 x？

像这样：

def task1(**context):
task_instance = context['task_instance']
task_instance.xcom_push(key="x", value="Hello")
generate_data = PythonOperator(
task_id="task1",
python_callable=task1,
dag=dag,
)
task2 = ECHOXOperator(x = task_instance.xcom_pull('task1', 'x'), provide_context=True)
task1 >> task2

这不起作用，因为未定义ECHOXOperator中的task_instance。

谢谢

Answer 1

您应该在自定义运算符中x作为templated_fields传递。

class ECHOXOperator(BaseOperator):
template_fields = ['x']
@apply_defaults
def __init__(self, x, *args, **kwargs):
self.x = x
super(ECHOXOperator, self).__init__(*args, **kwargs)
def execute(self, context):
print(self.x)

现在，您可以执行以下操作来获取在上一个任务中传递的 Xcom 值：

def task1(**context):
task_instance = context['task_instance']
task_instance.xcom_push(key="x", value="Hello")
generate_data = PythonOperator(
task_id="task1",
python_callable=task1,
dag=dag,
)
task2 = ECHOXOperator(x = "{{ ti.xcom_pull('task1', 'x') }}")
task1 >> task2

有关templated_fields和金贾模板的更多信息：https://airflow.readthedocs.io/en/latest/concepts.html#id1