Can't add external executors -> Plugin manager

[osallou]

Would be nice to be able to add other executors "out of airflow codebase".

List of executors is hard coded in airflow/executors/init.py

A kinda plugin mechanism could allow to add other executors.

[mistercrunch]

For operators, you can derive BaseOperator anywhere outside of the Airflow code and use them in your DAGs (we do that internally for operators that aren't relevant to the open source community).

For executors they are a little more deeply embedded in the code as you pointed out. If you need a hook in https://github.com/mistercrunch/airflow/blob/master/airflow/executors/__init__.py I'll be happy to accept it. Could be something like:

try:
    from airflow_custom import DEFAULT_EXECUTOR
except:
    DEFAULT_EXECUTOR = None

And then somehow have this take precedence over the if statements underneath.

Then all you need is a airflow_custom.py module in your environment that defines DEFAULT_EXECUTOR as a derivative of BaseExecutor.

[osallou]

To ease a community development of operators,i think that a plugin
mechanism such as provided by yappsy would be best. A plugin dir in AIRFLOW
dir could simply hold those airflow_XXX operators and init would simply
load the selected plugin (if not one of the base operator).

I can fork the project and propose such solution with a pull request if you
want.

Le sam. 6 juin 2015 00:21, Maxime Beauchemin [email protected] a
écrit :

For operators, you can derive BaseOperator anywhere outside of the Airflow
code and use them in your DAGs (we do that internally for operators that
aren't relevant to the open source community).

For executors they are a little more deeply embedded in the code as you
pointed out. If you need a hook in
https://github.com/mistercrunch/airflow/blob/master/airflow/executors/__init__.py
I'll be happy to accept it. Could be something like:

try:
from airflow_custom import DEFAULT_EXECUTOR
except:
DEFAULT_EXECUTOR = None

And then somehow have this take precedence over the if statements
underneath.

Then all you need is a airflow_custom.py module in your environment that
defines DEFAULT_EXECUTOR as a derivative of BaseExecutor.

—
Reply to this email directly or view it on GitHub
#3 (comment).

[mistercrunch]

Sounds perfect. Would the folder structure go $REPO/plugins/<plugin_set>/operators/<operator>.py or straight $REPO/plugins/operators/<operator>.py ? How should we enable plugins / plugin sets? Load all we find in the folders when present?

[nerdvegas]

+1 for plugin architecture via Yapsy.

[mistercrunch]

@osallou , I just read about Yapsy and this sounds like a great idea. We can squeeze a plugin_folder setting in airflow.cfg and operators / executors / macros in that folder would get discovered and integrated.

Internally we were mentioned the possibility of having plugins that would have UI components to them. Seems like it'd be doable too eventually.

[osallou]

I used Yapsy in several of my projects, it is really easy and you can group plugins by "type" (operators, executors, ...). Then you only need to match your config with available plugins.

Do you want me to code it and send a pull request or do you prefer to manage it yourself?

[osallou]

Regarding structure and Yapsy use, I think that all plugins could go directly in a plugin dir (defined in config or $airhome/plugins by default) then you load the one defined in config file (for operator).
With base objects, you can group plugins and load the expected one with something like

    # Build the manager
    simplePluginManager = PluginManager()
    # Tell it the default place(s) where to find plugins
    simplePluginManager.setPluginPlaces([plugins_dir_from_config])
    simplePluginManager.setCategoriesFilter({
       "Operator": BaseOperator,
       "Executor": BaseExecutor
     })
    # Collect all plugins
    simplePluginManager.collectPlugins()
    # Get an instance of the executor defined in config
    for pluginInfo in simplePluginManager.getPluginsOfCategory("Executor"):
       if pluginInfo.plugin_object.get_name() == executor_defined_in_config:
         self.executor = pluginInfo.plugin_object

[mistercrunch]

Sounds like we'd need a bit of code in operators/__init__.py to integrate the plugins that are instances of BaseOperator if we want them to be namespaced there. They could also be namespaced under airflow.plugins.operators.PluggedInOperator

I'm not sure how it's usually done, but it seems like it'd be nice to have them integrated in airflow.operators (same goes for executors and macros)

[codewithcheese]

A plugin system would be useful. I just wrote a hook and sensor operator for RabbitMQ so I could fire off a task when a queue became empty. master...codewithcheese:rabbitmq_hook_sensor

[mistercrunch]

I'm going to work on a plugin system over the next week, seems pretty straightforward. @codewithcheese, we support defining pool of tasks in Airflow, can your use case be handled by Airflow pools? http://pythonhosted.org/airflow/concepts.html#pools

[codewithcheese]

Actually I am using rabbitmq for data processing in a different project and need to run some bash commands when it is complete. I was sharing that to demonstrate that a plugin system for hooks and not so necessarily operators would be useful to me.

[mistercrunch]

I'm starting work on a plugin system using yapsy, I'll paste a link to the PR here when it's baked. I'm planning on integrating hooks, operators, macros, webviews, executors and I think that's it for now. We have use cases internally so that justifies the work.

[codewithcheese]

👍

[mistercrunch]

Merged 22ac771
Documented here: http://pythonhosted.org/airflow/plugins.html

Let me know what you think

[mistercrunch]

It's out on pypi (v1.1.0)

[osallou]

Sounds nice and fitting needs.
Thanks

Le mer. 17 juin 2015 03:09, Maxime Beauchemin [email protected] a
écrit :

Merged 22ac771
22ac771
Documented here: http://pythonhosted.org/airflow/plugins.html

Let me know what you think

—
Reply to this email directly or view it on GitHub
#3 (comment).

[codewithcheese]

Sweeet thanks!. ill try it out. Now i can stop maintaining a fork for my deployment.