GCP, Composer, Airflow, Operators

Question

As Google Cloud Composer uses Cloud Storage to store Apache Airflow DAGs. However, where the operators are stored ? I am getting an error as below:

Broken DAG: [/home/airflow/gcs/dags/example_pubsub_flow.py] cannot import name PubSubSubscriptionCreateOperator.

Answer 1

This operator was added in Airflow 1.10.0 . As of today, Cloud Composer is still using Airflow 1.9.0, hence this operator is not available yet. You can add this as a plugin.

Answer 2

apparently, according to the following post in this message in the Composer Google Group list, to install as a plugin the contrib is not needed to add the Plugin boilerplate. It is enough with registering the plugins via this command:

gcloud beta composer environments storage plugins import --environment dw --location us-central1 --source=custom_operators.py

See here for detail.

The drawback is that if your contrib operator uses others you will have to copy also those and modify the way they are imported in python, using:

from my_custom_operator import MyCustomOperator

instead of:

from airflow.contrib.operators.my_custom_operator import MyCustomOperator