使用 HttpSensor 触发 Airflow DAG答案

【问题标题】：Trigger Airflow DAG using HttpSensor使用 HttpSensor 触发 Airflow DAG
【发布时间】：2020-11-23 19:46:18
【问题描述】：

我正在尝试实现here 中提到的确切用例，但我遇到了问题。我似乎无法真正关闭触发器以执行任务t1，即使我正在更改引用的谷歌表。我已确保所有路径引用都是正确的，并且我可以访问必要的 API（谷歌驱动器）。我对 Airflow 很陌生，所以我怀疑我的 dag 缺少基本知识：

from airflow import DAG
from airflow.sensors.http_sensor import HttpSensor
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from datetime import timedelta, datetime

args = {
    'owner': 'airflow',
    'depends_on_past': False, 
    'start_date': datetime(2020, 11, 23),
    'email': 'my_email@gmail.com',
    'email_on_failure': True,
}

dag = DAG(
    'my_dag',
    default_args=args,
    description='blah blah',
    schedule_interval='*/5 * * * *'
)

# sensor to detect changes in google sheet
s = HttpSensor(
    task_id='check_sheet_change',
    http_conn_id='http_default',
   endpoint='https://www.googleapis.com/drive/v3/files/hiddenfromyouid',
    request_params={},
    response_check=lambda response: response.json()['modifiedTime'] > (datetime.now() - timedelta(seconds=30)),
    poke_interval=30,
    dag=dag,
)

# if s is triggered, do stuff in python
t1 = BashOperator(
    task_id='do stuff in python',
    bash_command='python3 /home/airflow/gcs/dags/pythons/mypyfile.py',
    dag=dag
)

s >> t1

我错过了什么？任何帮助表示赞赏。

【问题讨论】：

标签： python google-cloud-platform airflow

【解决方案1】：

endpoint of HttpSensor 用于 URL 的相对部分。

试试这个，http_default 连接应该直接https://www.googleapis.com

s = HttpSensor(
    task_id='check_sheet_change',
    http_conn_id='http_default',
   endpoint='/drive/v3/files/hiddenfromyouid',
    request_params={},
    response_check=lambda response: response.json()['modifiedTime'] > (datetime.now() - timedelta(seconds=30)),
    poke_interval=30,
    dag=dag,
)

【讨论】：