-----===== Airflow ROADMAP =====-----

[mistercrunch]

This github issue is meant to track the big roadmap items ahead of use and for the community to provide ideas, feedback, steer and comment

Lineage Annotations

Airflow knows a lot about your job dependencies, but how much does it know about your data objects (databases, tables, reports, ...)? Lineage annotations would expose a way for users to define lineage between data objects explicitly and tie that to tasks and/or DAG. The framework may include hooks for programatic inference (HiveOperator could introspect the HQL and guess table names and lineage), and a way to override or complement these. The framework will most likely be fairly agnostic about your data objects, letting you namespace them however you want, and simply consider them as an array of parent/child relationship between strings. It may be nice to use dot notation and reserve the first part of the expression to object_type, allowing for color coding in a graph view and tying actions (links) and the like. This will course ship with nice graph visualization, browsing and navigation features.

Picklin'

stateless/codeless web servers and workers

Backfilll UI

Trigger a backfill from the UI

REST API

Essentially offering features similar to what is available in the CLI through a REST API, there may be some automagic solutions here that can figure out how to build REST specs from argparse, it would also insure consistency between the two

[done!]

Continuous integration with Travis-CI

Systematically run all unit tests against CDH and HDP on Python 2.7 and 3.5

Externally Triggered DAGs

Airflow currently assumes that you run your workflows on a fixed schedule interval. This is perfect for hourly, daily and weekly jobs. When thinking about "analytics as a service" and "analysis automation", many use cases are more of the "on demand" nature. For instance if you have a workflow that processes someone's genome when ordered to, or a workflow that builds a dataset on demand for data scientists based on parameters they provide, .... This requires a new class of DAGs that are triggered externally, not on a fixed schedule interval.

[syvineckruyk]

👍

[kapil-malik]

Nice !
+1 for REST APIs, it will be helpful to have read/query APIs to list DAGs / jobs and their status etc.

Also, do you think it's possible that different steps of a DAG may have different resource requirements? Say for example memory ? I am a beginner in python, but I assume it would have a concept of specifying the requirements, just like we can do in java (via Xmx etc.).
Further, for distributed resource schedulers (Yarn / mesos / kubernetes / GCE etc.), even cpu cores / disk can be specified. Although I am afraid those specifications might not be applicable for local executors. So the behavior will differ with envts (rather undesirable).

One more thing, do you have plans for UI for defining a DAG? say some sort of drag drop which allows to select from available operators and allows for setting their properties (and for using xcom within DAG)

[syvineckruyk]

@kapil-malik be careful with mentioning drag and drop ... @mistercrunch has strong opinions about it :)

[kapil-malik]

Oh ok ..

Well I completely echo the sentiment behind python based authoring, I was only wondering if it'd add to adoption rate if people could manage e2e flow in UI (authoring, execution, logs). :)

[mistercrunch]

@kapil-malik about resources containment it will have to be some sort of integration with executors that support it. I had a brainstorm session talking about leveraging the JVM through using Jython only on the worker side for specific operators that would be stamped "jython compatible". I don't have a good sense on how complicated or feasible this is....

About the DAG generation from the UI I think I'd like to have something like that that would generate generate python code (to force the user to deal with the code and not bypass code review), and probably more form/CRUD based than drag/drop... It isn't a priority unless the community gets behind it.

[kapil-malik]

Yes, regarding resource requirements, I also thought whether it would be better to add them to BaseOperator, or have specific Operators. For example, Yarn / mesos executors could very well support this for say 'Docker Operators'.
While this might look too narrow / specific for original 'Airflow' users, I believe it will be extremely useful for original 'Yarn / Mesos / GCE' users, who are out there looking for a better workflow management solution. Yarn still has Oozie, but we have no such alternative in mesos (no offense to Chronos). Airflow can fill that space very nicely (and oh that would be such a relief), but by restricting to specific operators which take inputs more aligned with mesos.

[mistercrunch]

Clearly the best option is to add a parameter to BaseOperator that would be called executor_config or something, and the format is specified in the the executor's documentation. The format could be an object, a data structure, a reference to a server-side template, or whatever makes sense in the context of the executor.

Celery clearly has little or no support for this. At Airbnb we're planning on writing a YarnExecutor in the next few months.

[portah]

Hi,
Thank you for such great product as airflow.
I would like to speculate about "Externally Triggered DAGs". You've mentioned "genome", "data scientist" I'm working as bioinformatician for the last 4 years and I feel that useful, easy to install and configure workflow system is really the problem. So my speculation is about tools that bioinformatician usually run. The diversity of tools is incredible java/python/R/C++/perl based, some of them output the result into stdout some of them can put a bunch of files into current dir or relative to current subfolder. The names of output files can depend on input file/s name or be separated one. The organization of pipeline is not only the order of tools but also some logic, like, in one step we can mark some replicates on the other we can do analysis with or without replicates and based on result choose the best solution and produce output. Usually it is good to make different branches for later analysis if something wrong and repeat from specific position and downstream to save a time.

So my understanding to enable this kind of functionality and be useful airflow has to think about DAG like an instance but not each task and context has to be the same/shared for the whole DAG to pass all the data between steps.

So I just would like to have some insights about future logic for this functionality. To avoid going in wrong direction.

Thanks,
Andrey

[bolkedebruin]

Authenticated job submission?

[jlowin]

A few "smaller" ideas that I'd love to push forward. I'll add more as they occur to me.

1. Stateless DAGs

   DAGs live as pickled objects in the database, and are unpickled as needed. This would allow DAG .py scripts to live at any local or remote location since they only need to be read once in order to become available to the entire airflow ecosystem. If a user/worker/task notices or decides that a DAG pickle has expired, it can refresh the DAG for every consumer.

2. Improved worker logs

   We run airflow in a dockerized environment, where each worker is a separate microservice (and communicating across microservices is hard). As a result we have no way to access worker logs -- the workers don't communicate and even if they could, they could disappear/restart at any time. This isn't my area of expertise, but perhaps workers could write logfiles to a remote filesystem (S3?) rather than keeping them locally. Then the webserver could load and display them appropriately

With regard to ExternallyTriggeredDags -- here are some thoughts, some of which may make sense..

Today, DAGs are a hybrid of 1) task dependency graphs and 2) scheduler attributes (start_date, end_date, schedule_interval).

I suggest recasting a DAG as purely a dependency graph. It specifies the order in which tasks take place, but makes no claims about when they all start, stop, or repeat. It simply knows how to run its tasks in topological order, given an instruction to do so.

Those instructions could come from Scheduler objects which attach to DAGs and emit run times -- maybe just once, maybe on a regular schedule, maybe irregularly, maybe totally at random. When the attached Scheduler tells the DAG to run, it runs.

In other words, every DAG is really an "Externally Triggered DAG" -- it just happens that some of them are triggered on a regular schedule (and behave like DAGs do today). The Scheduler, instead of being a single master process that queries every DAG every 5 seconds to see if the DAG is supposed to run, becomes many smaller processes, one for each DAG, that actually kicks off the DAG at the right time. And the master process is nothing more than a Clock which tells each of the Schedulers that it's time to update...

Basically, I'm worried about having two classes of DAGs -- externally triggered vs "classic" -- when there really isn't a fundamental difference between them. Airflow 1.x is obviously designed around periodic DAG runs, so an externally triggered DAG represents something that needs special handling. But if Airflow 2.x is built around arbitrary run times, then these two DAG types could be unified very nicely. And I like that because they both have the same core purpose: run tasks in order.

[mistercrunch]

@jlowin , I like the clarity that breaking the schedule element away from the DAG brings. Breaking it apart means rethinking some of the UI design and things like SLA management, or just hiding some components where they don't make sense.

[georgebdavis]

One way to minimize change and keep value in UI would be to retain a
timestamp (time of request for on-demand) but introduce a user provided
label. Concatenating the two would give an ordered unique id to reference
runs even for near simultaneous requests, and sequenced UIs like the tree
view wouldn't change much.

[arijun]

I really like @jlowin's suggestion unify different methods of triggering DAGs. It might make sense to include these various triggers in the DAG though. Consider a case where I want to fire off a series of tasks when a file changes after 6pm. You could represent this either as a scheduler which is limited by FileSensor, or a file-watching ExternalTrigger which is limited by a time sensor. These would have very similar effects, but very different representations in the DAG with @jlowin's implementation.

An alternative might be to have a different class of node in the DAG that is a "trigger" node, which would take the place of that separately defined scheduler. Thus the first node of a DAG would (almost) always be one of these "trigger" nodes, like a TimeOfDay trigger for daily tasks or FileUpdated trigger for on demand tasks.

These trigger nodes would be fundamentally different than Operator nodes in that they would be stateful--they would fire only once, each time their requirements are met. For example, a TimeOfDay trigger would conceptually differ from a TimeSensor operator in the sense that

• if you ask a TimeSensor directly upstream of you if you can run the answer will always be yes (assuming the time has passed)
• if you ask a TimeOfDay trigger directly upstream of you if you can run, the answer will be yes only for one run. If you do a run right after the time of day passes, asking again will give you a no.

Conceptually, this is very similar to dataflow programming languages like LabView, and with good reason. I think that those kind of languages provide a very fruitful analogue to what people mean in describing their workflow. To paraphrase this stack overflow answer, airflow could be seen as a state of 'this is running now' flowing through otherwise static instructions (a DAG), like how electrical signals flow through circuits or water flows through pipes. In this view, the triggers would be the generators of the 'this is running now' state.

[mistercrunch]

@arijun, interesting points. Triggers remind me of an oscillators in a synthesizer. Now if this is incorporated inside a DAG, that means that the triggering element acts on all (and only) downstream tasks from that point.

Related notes, and more related to having a trigger at a DAG level: the current primary key for a task instance is (dag_id, task_id, execution_date), I was thinking in sticking with that, but having an external notion of which execution_dates should actually exist and a mean of attaching a run_id of some kind. You can picture a new model DagRun with (dag_id, execution_date, run_id), it defines the schedules that should exist for each DAG, the trigger can label that run using the run_id for future references. For a given scheduled DAG, the scheduler would probably insert new dag schedules in this model, and for externally triggered dags, an event of some kind would insert a new timestamp.

[georgebdavis]

@mistercrunch $0.02 on the ID scheme (which makes good sense to me - no reason to forego the timestamp, which is useful even if something else triggered the run!): either make run_id a user-defined string or make sure there's a way to associate user tags with a run_id. This snags you two short term benefits and a third in the distance:

1. CLI commands will be easier to use if you can reference your own label for a run. In cases where runs are sparse in time, it might be sufficient to specify just the user-defined label and avoid having to remember the date you ran it. (E.g. airflow clear customer_cancellation_analysis generate_report -label customer_user_name, without having to remember what date you generated the analysis.)
2. If the label is accessible to DAG tasks (e.g. via the context), it could be a way to pass some metadata into a run (e.g. parameters of an on-demand run, like the above customer_user_name).
3. In the future, labels might let you subselect runs for some of the same views that airflow provides now. Imagine a DAG that analyzes stats within an availability zone in your architecture, and you trigger it once per day per AZ with the label for that AZ. A new dropdown on the treeview could let you look at just the history for one AZ.

Some of the use cases above might be better served by specific run_id-associated metadata, but just having one user-settable label would give a head start on all of them, and probably stay useful even if the metadata gets added at some future point.