气流S3键传感器-如何使其继续运行

在…的帮助下 this Stackoverflow post 我刚做了一个程序(如文章所示),当一个文件放在S3桶中时,一个正在运行的DAG中的任务被触发,然后我使用bashoperator执行一些工作。一旦完成,尽管DAG不再处于运行状态,而是进入成功状态,如果我想让它获取另一个文件,我需要清除所有“过去”、“未来”、“上游”、“下游”活动。我想做这个程序,这样它总是在运行,每当一个新的文件放在S3存储桶中,程序就开始执行任务。

我可以继续使用s3keysenor来完成这个任务吗,或者我需要找出一种设置 External Trigger 来运行我的DAG?到目前为止,如果只运行一次,我的S3keysensor就没有意义了。

from airflow import DAG
from airflow.operators import SimpleHttpOperator, HttpSensor, EmailOperator, S3KeySensor
from datetime import datetime, timedelta
from airflow.operators.bash_operator import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2018, 5, 29),
    'email': ['something@here.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 5,
    'retry_delay': timedelta(minutes=5)
}

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

# This Activity runs a Python script that creates an AWS EMR cluster and then does EMR activity on the EMR cluster.
t2 = BashOperator(
    task_id='create_emr_cluster_1',
    bash_command='python /home/ec2-user/aws-python-sample/Create_EMR_Then_Do_EMR_Activities.py',
    retries=1,
    dag=dag)

t1 = BashOperator(
    task_id='success_log',
    bash_command='echo "Dag ran successfully" >> /home/ec2-user/s3_triggered_dag.txt',
    dag=dag)

sensor = S3KeySensor(
    task_id='new_s3_file_in_foobar-bucket',
    bucket_key='*',
    wildcard_match=True,
    bucket_name='foobar-bucket',
    s3_conn_id='s3://foobar-bucket',
    timeout=18*60*60,
    poke_interval=120,
    dag=dag)

t1.set_upstream(sensor)
t2.set_upstream(t1)

我想知道这是否不可能,因为它不是有向非循环图,而是有一个重复的循环 传感器->T1->T2->传感器->T1->T2->传感器->…不断重复 。

更新:

我的用例非常简单,每当一个新文件被放置在一个指定的AWS S3桶中时,我希望触发DAG并开始我的各种任务的处理。这些任务将执行如下操作:实例化一个新的AWS EMR集群,从AWS S3存储桶中提取文件,执行一些AWS EMR活动,然后关闭AWS EMR集群。从那里,DAG将返回到等待状态,在该状态下,它将等待新文件到达AWS S3存储桶,然后无限期地重复该过程。