Airflow Foreach + Sensors

[valayDave]

Stacked On Top of #1224

This PR enabled Airflow Sensor Support and Airflow Foreach Support.

Support for following Airflow sensor decorators:

• @airflow_external_task_sensor
• @airflow_s3_key_sensor

@airflow_s3_key_sensor

Sensor Args

The below flow will trigger when Airflow detects a x.json is present in the S3 Bucket path

from metaflow import FlowSpec, step
from metaflow import  airflow_s3_key_sensor

SENSOR_PATH = "s3://my-s3-bucket/sensor-data/x.json"

@airflow_s3_key_sensor(bucket_key=SENSOR_PATH,mode='poke')
class SensorFlow(FlowSpec):
        
        @step
        def start(self):
            self.x = 1
            self.next(self.b)

        @step
        def b(self):
            self.x += 1
            self.next(self.end)
        
        @step
        def end(self):
            self.x += 1
            print("Value of X is ", self.x)
    
if __name__ == "__main__":
    SensorFlow()

@airflow_external_task_sensor

Sensor Args

The below flow will get triggered when SourceDag-0 and SourceDag-1 with the same execution_date finish completion.

from metaflow import FlowSpec, step
from metaflow import  airflow_external_task_sensor, schedule

@airflow_external_task_sensor(name='x',external_dag_id='SourceDag-0',mode='poke')
@airflow_external_task_sensor(name='y',external_dag_id='SourceDag-1',mode='poke')
@schedule(cron='* * * * *')
class SensorFlow(FlowSpec):
        
        @step
        def start(self):
            self.x = 1
            self.next(self.b)

        @step
        def b(self):
            self.x += 1
            self.next(self.end)
        
        @step
        def end(self):
            self.x += 1
            print("Value of X is ", self.x)
    
if __name__ == "__main__":
    SensorFlow()

[romain-intel]

I did not review anything in airflow/. This needs to be rebased on 1212 prior to merging.

[romain-intel]

Did not review airflow specific parts but rest looks fine.