I need a flow to start with a sensor

[r39132]

I have a flow that is linear and sequential as shown below :

1. new data arrives in SQS
2. wait for that data to start being written to postgres
3. wait for the SQS queue to drain
4. send an email that the flow completed

Eventually, we will have a pre-cursor step (e.g. step 0.) that kicks off a Spark job before waiting for new data in the SQS queue. For now, I do not have that, so I would like my flow to simply reset to the first step. In some ways, I am asking for a circular flow that never ends. This breaks the entire workflow abstraction. On another note, where is the "cron" functionality? Do I schedule airflow commands in cron outside of airflow?

[mistercrunch]

Airflow runs its own scheduler, it's hard to miss it in the docs:
http://pythonhosted.org/airflow/scheduler.html

Is your process a batch job that runs on a fixed scheduled interval? If not Airflow is probably not the right platform. DAG stands for Directed Acyclic Graph. Acyclic means there's no loop. Though its assumed we run the same set of jobs (the DAG) at a scheduled interval, which kind of makes for a large loop around the DAG execution.

[r39132]

Hi!
Yes, it is batch oriented. I'm waiting for some changes in how we submit spark jobs so that I can trigger my flow starting at the Spark job submission. I've covered 4/5 stages of my flow, unfortunately, they are stages 2-5, inclusive.

Yes, the large loop is triggered by a (daily?) schedule. What would be nice would be to have the large loop triggered by different signals. Instead of by a clock (interrupt), by a message on an SQS queue.

This is very whimsical and nothing urgent.

[r39132]

BTW, having worked with Azkaban for 3 years at LinkedIn, I really like the Python DSL approach. I think it lends itself to more possibilities than just declaratively building a DAG of execution that is triggered by a cron.

[r39132]

About the scheduler and cron, I'll clarify. I was expecting to find a calendar widget - call it UI sugar.

In reading #7, it seems that this is instead defined in the python DAG definition code. And essentially, in order to alter the schedule of an existing DAG, we need to alter and re-run the DAG code. So, in a production environment, people are updating their DAG code on some production machine just to change the frequency of their running jobs, which can be problematic if that engineer introduced more changes than just an interval change. Hence, a calendar widget provides the option for limited scope and ease of change.

[mistercrunch]

I haven't thought of frequency of execution as a variable, I always thought of it as a constant. I thought it could evolve to become a more complex constant by eventually adding a feature supporting the cron syntax (5 * * * *) to allow for more complex rules of execution.

Though it seems you are interested in a more on-demand or on-event approach.

In Airflow, a task instance is essentially a specific task run (the unit at which we track state) and this is done by associating a timestamp to the task definition. The scheduler is aware of the DAG's schedule and triggers based on that.

I guess the notion of task or DAG can be de-coupled from its schedule and can be triggered on demand using the API or the CLI, externally.

The on-demand approach seems somewhat easy to integrate to the platform, where you'd need a bit of metadata saying I want this DAG to run for this date or timestamp and the scheduler would pick that up and make sure that tasks get triggered.

The on-event would require some sort of timeless event listener that could leverage the on-demand infrastructure to add an entry to the on-demand list.

It's flipping the current approach around: a fixed schedule with sensors in it VS a sensor that defines the schedule

[r39132]

We integrated Spark with my main flow, so I now have an end-to-end flow that works, with a few rough-edges remaining.

[pedrorjbr]

I need this feature too! I would like to kick off a DAG when a file is upload to S3 bucket. Is anyone has a solution for on-event approach?

[bolkedebruin]

No solution yet. Just a thought. With a Notify daemon (to be implemented) that does polling and puts actions into the executor queues when an event happens. lib event (python-libevent) comes to mind

[pedrorjbr]

I used a S3 lambda function to trigger an event that starts a docker container in the AWS ECS infrastructure. So I did not used the Airflow :(

[mistercrunch]

Since PR #540 we support external triggers for DAGs, meaning you can order a DAG run externally, not on a schedule.

The current way to trigger is by using the CLI airflow trigger_dag dag_id. I'm planning on adding a DagRunTrigger operator that receives a function that if it returns an non-None object creates a DagRun for a specified dag_id, the returned object is pickled to the database and made available to the context of all tasks in this DagRun.