Cloud Composer Dependency Issue (June 2018)

foo_dag.py

import airflow
from airflow.operators.bash_operator import BashOperator
from airflow.models import DAG

from foo_dep import print_something

args = {"owner": "airflow", "start_date": airflow.utils.dates.days_ago(2)}
dag = DAG(dag_id="foo", default_args=args, schedule_interval="0 0 * * *")
a_bash_cmd = BashOperator(task_id="a_bash_cmd", bash_command="echo 1", dag=dag)

if __name__ == "__main__":
    dag.cli()

foo_dep.py

def print_something():
    print "Something."

Upload both of the above to your Cloud Composer bucket under dags/.

The DAG will be broken because it can't load the dependency:

This works locally, however.

Create a subfolder and have all dependent modules in that. Bundle everything as zip file and upload to dag folder. https://airflow.apache.org/concepts.html#packaged-dags

Thanks!

For posterity, create packages instead of modules at the top-level, like so:

foo_dag.py
foo_dep/__init__.py
foo_dep/foo_dep.py

Make sure foo_dag imports the dependency correctly. The Gist above won't work, but from foo_dep.foo_dep import print_something will.

Now, zip that folder structure and upload it to your /dags folder on GCS. Your Airflow environment will pick it up.