-
Notifications
You must be signed in to change notification settings - Fork 14.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
A simple plugin system for Airflow #32
Merged
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,58 @@ | ||
import imp | ||
import inspect | ||
import logging | ||
import os | ||
|
||
from airflow.configuration import conf | ||
|
||
|
||
plugins_folder = conf.get('core', 'plugins_folder') | ||
if not plugins_folder: | ||
plugins_folder = conf.get('core', 'airflow_home') + '/plugins' | ||
plugins_folder = os.path.expanduser(plugins_folder) | ||
|
||
plugin_modules = [] | ||
# Crawl through the plugins folder to find pluggable_classes | ||
templates_dirs = [] | ||
|
||
for root, dirs, files in os.walk(plugins_folder): | ||
if os.path.basename(root) == 'templates': | ||
templates_dirs.append(root) | ||
for f in files: | ||
try: | ||
filepath = os.path.join(root, f) | ||
if not os.path.isfile(filepath): | ||
continue | ||
mod_name, file_ext = os.path.splitext( | ||
os.path.split(filepath)[-1]) | ||
if file_ext != '.py': | ||
continue | ||
m = imp.load_source(mod_name, filepath) | ||
plugin_modules.append(m) | ||
except Exception() as e: | ||
logging.exception(e) | ||
logging.error('Failed to import plugin ' + filepath) | ||
|
||
|
||
def register_templates_folders(app): | ||
from jinja2 import ChoiceLoader, FileSystemLoader | ||
new_loaders = [FileSystemLoader(s) for s in templates_dirs] | ||
app.jinja_env.loader = ChoiceLoader([app.jinja_env.loader] + new_loaders) | ||
|
||
|
||
def get_plugins(baseclass, expect_class=True): | ||
""" | ||
Set expect_class=False if you want instances of baseclass | ||
""" | ||
# Late Imports to aoid circular imort | ||
plugins = [] | ||
for m in plugin_modules: | ||
for obj in m.__dict__.values(): | ||
if (( | ||
expect_class and inspect.isclass(obj) and | ||
issubclass(obj, baseclass) and | ||
obj is not baseclass) | ||
or | ||
(not expect_class and isinstance(obj, baseclass))): | ||
plugins.append(obj) | ||
return plugins |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -35,4 +35,5 @@ Content | |
profiling | ||
cli | ||
scheduler | ||
plugins | ||
code |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,101 @@ | ||
Plugins | ||
======= | ||
|
||
Airflow has a simple plugin manager built-in that can integrate external | ||
features at its core by simply dropping files in your | ||
``$AIRFLOW_HOME/plugins`` folder. | ||
|
||
The python modules in the ``plugins`` folder get imported, | ||
and **hooks**, **operators**, **macros**, **executors** and web **views** | ||
get integrated to Airflow's main collections and become available for use. | ||
|
||
Objects | ||
------- | ||
|
||
* Classes derived from ``BaseOperator`` land in ``airflow.operators`` | ||
* Classes derived from ``BaseHook`` land in ``airflow.hooks`` | ||
* Classes derived from ``BaseExecutor`` land ``airflow.executors`` | ||
* object created from a class derived from ``airflow.PluginView`` get integrated in the Flask app | ||
* object created from ``AirflowMacroPlugin(namespace="foo")`` land in ``airflow.macros.foo`` | ||
* Files located in subfolders named ``templates`` folders become available when rendering pages | ||
* (upcoming) CLI subcommands | ||
|
||
|
||
What for? | ||
--------- | ||
|
||
Airflow offers a generic toolbox for working with data. Different | ||
organizations have different stacks and different needs. Using Airflow | ||
plugins can be a way for companies to customize their Airflow installation | ||
to reflect their ecosystem. | ||
|
||
Plugins can be used as an easy way to write, share and activate new sets of | ||
features. | ||
|
||
There's also a need for a set of more complex application to interact with | ||
different flavors of data and metadata. | ||
|
||
Examples: | ||
|
||
* A set of tools to parse Hive logs and expose Hive metadata (CPU /IO / phases/ skew /...) | ||
* An anomaly detection framework, allowing people to collect metrics, set thresholds and alerts | ||
* An auditing tool, helping understand who accesses what | ||
* A config-driven SLA monitoring tool, allowing you to set monitored tables and at what time | ||
they should land, alert people and exposes visualization of outages | ||
* ... | ||
|
||
|
||
Why build on top Airflow? | ||
------------------------- | ||
|
||
Airflow has many components that can be reused when building an application: | ||
|
||
* A web server you can use to render your views | ||
* A metadata database to store your models | ||
* Access to your database, and knowledge of how to connect to them | ||
* An array of workers that can allow your application to push workload to | ||
* Airflow is deployed, you can just piggy back on it's deployment logistics | ||
* Basic charting capabilities, underlying libraries and abstractions | ||
|
||
|
||
Example | ||
------- | ||
|
||
The code bellow defines a plugin that injects a set of dummy object | ||
definitions in Airflow. | ||
|
||
.. code:: python | ||
|
||
# Importing base classes that we need to derive | ||
from airflow.hooks.base_hook import BaseHook | ||
from airflow.models import BaseOperator | ||
from airflow.executors.base_executor import BaseExecutor | ||
from airflow import AirflowViewPlugin, AirflowMacroPlugin | ||
from flask_admin import expose | ||
|
||
# Will show up under airflow.hooks.PluginHook | ||
class PluginHook(BaseHook): | ||
pass | ||
|
||
# Will show up under airflow.operators.PluginOperator | ||
class PluginOperator(BaseOperator): | ||
pass | ||
|
||
# Will show up under airflow.executors.PluginExecutor | ||
class PluginExecutor(BaseExecutor): | ||
pass | ||
|
||
# Shows up in the UI in menu -> Plugins -> Test | ||
class TestView(AirflowViewPlugin): | ||
@expose('/') | ||
def query(self): | ||
return self.render("test.html", content="Hello galaxy!") | ||
v = TestView(category="Plugins", name="Test") | ||
|
||
|
||
# Available as other macros in templates {{ macros.foo_plugin.success() }} | ||
def success(): | ||
return "Success!" | ||
obj = AirflowMacroPlugin(namespace="foo_plugin") | ||
obj.success = success | ||
|
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Maybe minor bump in the version?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I bump as part of the process when I decide to upload to pypi