我正在学习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 API中的一些端点发出了两个GET请求。目标是学习如何通过Airflow处理一些检索到的数据。
执行很顺利-我的代码成功地调用了任务t1和t2中的端点,我可以在Airflow UI中看到它们按照我编写的
set_upstream
规则的正确顺序被记录下来。我无法弄清楚的是如何访问这两个任务的JSON响应。看起来很简单,但我无法弄清楚。在SimpleHtttpOperator中,我看到一个response_check参数,但没有任何东西可以简单地打印、存储或查看JSON响应。
谢谢。
t1
的响应中访问任务t2
中的数据?如果你能分享这个信息就太好了。Chengzhi的答案讲解了如何推送和获取,但如何在任务t2
中拉取呢? - Jacobian