Streamlined (Functional) Airflow [ARCHIVED]

To do

• Revise start_date
  •  optional for DAGs
  •  remove for Tasks
•  Add sensible defaults for all parameters (Done: owner=Airflow)
•  Autogenerate task_ids?
•  Defer/infer DAG membership #1318
• Syntactic sugar
  •  @decorator syntax for creating tasks
  •  DAGs as context managers #1318
  •  "bitshift" (>>, <<) chain syntax #1318
  •  "call" chain syntax? (e.g. like "fairflow")

Objective

Minimize required arguments for creating DAGs and tasks by auto-generating or inferring them where possible. Take advantage of Python syntax and processing to streamline code.

Why?

Because Airflow's strength is "workflows as code" and the more streamlined we can make that code, the better.

Because having many repeated [and unnecessary] parameters makes code unwieldy and fragile.

Because even the simplest Airflow workflows require a lot of boilerplate setup, and that's an impediment to easy adoption and widespread use. And complex workflows can get lost behind all the code and repeated declarations. Compared to many workflow managers, Airflow code is easy to grok. But for someone without knowledge of Airflow, it's still hard to understand what's going on.

How?

• defer or infer task parameters that are currently required at task creation
  • Make start_date completely optional for DAGs
    • if available, don't run tasks prior to start_date
    • otherwise, I'm not sure it matters
  • Formally remove start_date from tasks
    • I think it's already disregarded but still has to be set
  • Sensible defaults
    • default owner: 'Airflow'
    • default interval: 1 day (already happens)
  • Auto-generate task_ids if not provided
    • task_id := class name + unique hash (or int)
    • users can still supply task_id if they want
      • in fact maybe task_id is always auto-generated and users supply a name ordisplay_name. This would have a nice unification where task and dag would both have a .name property (today must call either task_id or dag_id as appropriate)
• Infer DAG membership
  • Tasks do not have to be assigned a DAG when they are created.
  • If a task without a DAG is connected (upstream/downstream) to a task with a DAG, then it adopts that DAG as its own.
  • If tasks are connected with conflicting DAGs, raise an error
• syntactic sugar via Python
  • @decorators
    • quickly transform functions into Tasks
      • obvious to do this with PythonOperators but could also do with BashOperator(or any other)
      • optional decorator arguments to supply optional information like task_id, dag,upstream, etc.
  • context managers
    • DAGs can be used as context managers
    • with dag: any tasks created in task manager are applied to that dag (see example)
  • "pipe" syntax to chain tasks (more "unix")
    • workflow = upstream_task | downstream_task
    • Advantage: can easily chain multiple tasks together
  • "call" syntax to chain tasks (more "pythonic")
    • workflow = downstream_task(upstream_task)
    • Advantage: looks like python!
    • >>> a __call__ pattern can be added on top of existing Airflow using a dictionary of task instances. See https://github.com/michaelosthege/fairflow for details

Illustrative Example

Typical [wordy] setup:

dag = airflow.DAG(
    dag_id='my_dag',
    default_args=dict(
        owner='jlowin',
        start_date=datetime(2015, 1, 1)
  )
)

def fn_1():
  msg = "Hello, world!"
  print(msg)
  return msg

op_1 = airflow.PythonOperator(
    task_id='op_1',
    dag=dag,
    python_callable=fn_1
)

def fn_2():
  msg = "Goodbye, world!"
  print(msg)
  return msg

op_2 = airflow.PythonOperator(
    task_id='op_2',
    dag=dag,
    python_callable=fn_2
)

op_3 = airflow.BashOperator(
    task_id='op_3',
    dag=dag,
    bash_command='echo "Hello from bash, world!"'
)

op_1.set_downstream(op_2)
op_2.set_downstream(op_3)

Proposed [streamlined] setup of the same workflow. This extreme case is totally over the top, just trying to show lots of ideas at once:

# create the dag
dag = airflow.DAG('my_dag')

# use dag as a context manager
with dag:

    @airflow.task
    def fn_1():
      msg = "Hello, world!"
      print(msg)
      return msg

    @airflow.task
    def fn_2():
      msg = "Goodbye, world!"
      print(msg)
      return msg

    @airflow.bash_task(upstream=fn_2)
    def fn_3():
      'echo "Hello from bash, world!"'

    # two ways to set dependencies (other than decorator arguments)
    fn_1 | fn_2
    # OR
    fn_2(fn_1)