airflow - 如何访问来自 Airflow SimpleHttpOperator GET 请求的响应

Question

我正在学习 Airflow 并有一个简单的问题。下面是我的 DAG，名为 dog_retriever :

import airflow
from airflow import DAG
from airflow.operators.http_operator import SimpleHttpOperator
from airflow.operators.sensors import HttpSensor
from datetime import datetime, timedelta
import json



default_args = {
    'owner': 'Loftium',
    'depends_on_past': False,
    'start_date': datetime(2017, 10, 9),
    'email': 'rachel@loftium.com',
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=3),
}

dag = DAG('dog_retriever',
    schedule_interval='@once',
    default_args=default_args)

t1 = SimpleHttpOperator(
    task_id='get_labrador',
    method='GET',
    http_conn_id='http_default',
    endpoint='api/breed/labrador/images',
    headers={"Content-Type": "application/json"},
    dag=dag)

t2 = SimpleHttpOperator(
    task_id='get_breeds',
    method='GET',
    http_conn_id='http_default',
    endpoint='api/breeds/list',
    headers={"Content-Type": "application/json"},
    dag=dag)
    
t2.set_upstream(t1)

作为测试 Airflow 的一种方法，我只是在这个非常简单的 http://dog.ceo 中向某些端点发出两个 GET 请求。应用程序接口(interface)。目标是学习如何处理通过 Airflow 检索到的一些数据
执行正常 - 我的代码成功调用了任务 t1 和 t2 中的端点，我可以看到它们以基于 set_upstream 的正确顺序记录在 Airflow UI 中。我写的规则。
我无法弄清楚的是如何访问这两个任务的 JSON 响应。这看起来很简单，但我无法弄清楚。在 SimpleHtttpOperator 中，我看到 的参数response_check ，但不能简单地打印、存储或查看 JSON 响应。
谢谢。

Answer 1

因此，由于这是 SimpleHttpOperator 并且实际的 json 被推送到 XCOM，您可以从那里获取它。这是该操作的代码行:https://github.com/apache/incubator-airflow/blob/master/airflow/operators/http_operator.py#L87

您需要做的是设置xcom_push=True ，因此您的第一个 t1 将如下所示:

t1 = SimpleHttpOperator(
    task_id='get_labrador',
    method='GET',
    http_conn_id='http_default',
    endpoint='api/breed/labrador/images',
    headers={"Content-Type": "application/json"},
    xcom_push=True,
    dag=dag)

您应该能够找到所有带有 return value 的 JSON在 XCOM 中，可以在以下位置找到 XCOM 的更多详细信息: https://airflow.incubator.apache.org/concepts.html#xcoms