Wanted to learn airflow and using it through docker.
I used the LocalExecutor compose file with uncommented volume for the plugin folder.
Here my project structure :
โโโ dags
โย ย โโโ dag.py
โโโ docker-compose.yml
โโโ plugins
โโโ __init__.py
โโโ custom_plugins.py
โโโ operators
โโโ __init__.py
โโโ facts_calculator.py
โโโ has_rows.py
โโโ s3_to_redshift.py
I try then to follow the official documentation but I cannot import my custom operator.
Here the content of my custom_plugin.py (I tried to put it in the __init__ file too). I cannot import from operators only without an error.
from airflow.plugins_manager import AirflowPlugin
from plugins.operators.facts_calculator import FactsCalculatorOperator
from plugins.operators.has_rows import HasRowsOperator
from plugins.operators.s3_to_redshift import S3ToRedshiftOperator
# Defining the plugin class
class CustomPlugin(AirflowPlugin):
name = "custom_plugin"
# A list of class(es) derived from BaseOperator
operators = [
FactsCalculatorOperator,
HasRowsOperator,
S3ToRedshiftOperator
]
# A list of class(es) derived from BaseHook
hooks = []
# A list of class(es) derived from BaseExecutor
executors = []
# A list of references to inject into the macros namespace
macros = []
# A list of objects created from a class derived
# from flask_admin.BaseView
admin_views = []
# A list of Blueprint object created from flask.Blueprint
flask_blueprints = []
# A list of menu links (flask_admin.base.MenuLink)
menu_links = []
And in my /dags folder I do :
from airflow.operators import HasRowsOperator
from airflow.operators import S3ToRedshiftOperator
...
But I got an error from my IDE, if I try to run it I got this error:
[2019-10-06 15:34:03,054] {{dagbag.py:205}} ERROR - Failed to import: /usr/local/airflow/dags/dag.py
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 202, in process_file
m = imp.load_source(mod_name, filepath)
File "/usr/local/lib/python3.7/imp.py", line 171, in load_source
module = _load(spec)
File "<frozen importlib._bootstrap>", line 696, in _load
File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 728, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "/usr/local/airflow/dags/dag.py", line 5, in <module>
from airflow.operators import HasRowsOperator
ImportError: cannot import name 'HasRowsOperator' from 'airflow.operators' (unknown location)
It has been 2 days that I'm block here. I tried different solution found here and there but I cannot make it work. Any idea how to fix it ? Any exemple, even with a dummy operator ?
EDIT :
I tried to find online working exemple and I found this one : https://github.com/blue-yonder/airflow-plugin-demo
I added the compose file without touching anything else and I also have the same issue. So I don't know where this come from but it's hard to me to belive that there is not something wrong with the docker image.
I had a lot of issues getting plugins to work recently. The directory structure that did the trick was:
โโโ dags
โ โโโ dag.py
โโโ docker-compose.yml
โโโ plugins
โโโ something_plugin
โโโ __init__.py
โโโ operators
โโโ __init__.py
โโโ something_operator.py
Empty plugins/something_plugin/operators/__init__.py file.
For the plugins/something_plugin/__init__.py, something like:
from airflow.plugins_manager import AirflowPlugin
from plugin_folder.operators.something_operator import SomethingOperator
# Defining the plugin class
class SomethingPlugin(AirflowPlugin):
name = "something_plugin"
# A list of class(es) derived from BaseOperator
operators = [
SomethingOperator
]
# A list of class(es) derived from BaseHook
hooks = []
# A list of class(es) derived from BaseExecutor
executors = []
# A list of references to inject into the macros namespace
macros = []
# A list of objects created from a class derived
# from flask_admin.BaseView
admin_views = []
# A list of Blueprint object created from flask.Blueprint
flask_blueprints = []
# A list of menu links (flask_admin.base.MenuLink)
menu_links = []
And for importing in a DAG,
from airflow.operators.something_plugin import SomethingOperator
Current version of Airflow should warn if you try to import from airflow.operators instead of airflow.operators.thing_operator for builtin operators (as well as hooks, etc.) In this case, from your plugin instead of from the relevant file. Hooks would be the same.
from airflow.hooks.something_plugin import SomethingHook
Most helpful comment
I had a lot of issues getting plugins to work recently. The directory structure that did the trick was:
Empty
plugins/something_plugin/operators/__init__.pyfile.For the
plugins/something_plugin/__init__.py, something like:And for importing in a DAG,
Current version of Airflow should warn if you try to import from
airflow.operatorsinstead ofairflow.operators.thing_operatorfor builtin operators (as well as hooks, etc.) In this case, from your plugin instead of from the relevant file. Hooks would be the same.