Renaming project from Flux to Airflow

This commit is contained in:
Maxime Beauchemin committed 2014-10-14 18:28:29 -07:00
1 parent 323571ebd1
commit 88c99624f3
58 files changed
+164 -165

No files matched your search

+2 -2
View File
@@ -1,5 +1,5 @@
Flux
Airflow
====
Flux is a system to programmaticaly author, schedule and monitor data pipelines.
Airflow is a system to programmaticaly author, schedule and monitor data pipelines.
[Documentation](https://readthedocs.org/dashboard/flux-data-pipelines/)
File renamed without changes.
+5 -6
View File
@@ -1,8 +1,7 @@
#!/usr/bin/env python
from flux import settings
from flux.models import DagBag, TaskInstance, DagPickle, State
from sqlalchemy.ext.declarative import declarative_base
from airflow import settings
from airflow.models import DagBag, TaskInstance, DagPickle, State
import dateutil.parser
from datetime import datetime
@@ -185,7 +184,7 @@ def initdb(args):
"Proceed? (y/n)").upper() == "Y":
logging.basicConfig(level=logging.DEBUG)
from flux import models
from airflow import models
logging.info("Dropping tables that exist")
models.Base.metadata.drop_all(settings.engine)
@@ -198,8 +197,8 @@ def initdb(args):
session.query(models.DatabaseConnection).delete()
mysqldb = models.DatabaseConnection(
db_id='local_mysql', db_type='mysql',
host='localhost', login='flux', password='flux',
schema='flux')
host='localhost', login='airflow', password='airflow',
schema='airflow')
session.add(mysqldb)
session.commit()
File renamed without changes.
@@ -1,8 +1,8 @@
import logging
import time
from flux import settings
from flux.utils import State
from airflow import settings
from airflow.utils import State
class BaseExecutor(object):
@@ -74,7 +74,7 @@ class LocalWorker(multiprocessing.Process):
print command
command = (
"exec bash -c '"
"cd $FLUX_HOME;\n" +
"cd $AIRFLOW_HOME;\n" +
"source init.sh;\n" +
command +
"'"
@@ -91,7 +91,7 @@ class LocalWorker(multiprocessing.Process):
class LocalExecutor(BaseExecutor):
def __init__(self, parallelism=16):
def __init__(self, parallelism=8):
super(LocalExecutor, self).__init__()
self.parallelism = parallelism
File renamed without changes.
File renamed without changes.
File renamed without changes.
@@ -1,7 +1,7 @@
import logging
import sys
import os
from flux import settings
from airflow import settings
# Adding the Hive python libs to python path
sys.path.insert(0, settings.HIVE_HOME_PY)
@@ -13,7 +13,7 @@ from thrift.protocol import TBinaryProtocol
from hive_metastore import ThriftHiveMetastore
from hive_service import ThriftHive
from flux.hooks.base_hook import BaseHook
from airflow.hooks.base_hook import BaseHook
METASTORE_THRIFT_HOST = "localhost"
METASTORE_THRIFT_PORT = 10000
@@ -1,6 +1,6 @@
import MySQLdb
from flux import settings
from flux.models import DatabaseConnection
from airflow import settings
from airflow.models import DatabaseConnection
class MySqlHook(object):
File renamed without changes.
+6 -6
View File
@@ -15,10 +15,10 @@ from sqlalchemy import (
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from flux import macros
from flux.executors import DEFAULT_EXECUTOR
from flux import settings
from flux import utils
from airflow import macros
from airflow.executors import DEFAULT_EXECUTOR
from airflow import settings
from airflow import utils
from settings import ID_LEN
import socket
from utils import State
@@ -190,7 +190,7 @@ class TaskInstance(Base):
force=False,
pickle=None):
"""
Returns a command that can be executed anywhere where flux is
Returns a command that can be executed anywhere where airflow is
installed. This command is part of the message sent to executors by
the orchestrator.
"""
@@ -203,7 +203,7 @@ class TaskInstance(Base):
if not pickle and self.task.dag and self.task.dag.filepath:
subdir = "-sd {0}".format(self.task.dag.filepath)
return (
"flux run "
"airflow run "
"{self.dag_id} {self.task_id} {iso} "
"{mark_success} "
"{pickle} "
File renamed without changes.
@@ -2,9 +2,9 @@ from datetime import datetime
import logging
from time import sleep
from flux import settings
from flux.models import BaseOperator, TaskInstance, State
from flux.hooks import MySqlHook
from airflow import settings
from airflow.models import BaseOperator, TaskInstance, State
from airflow.hooks import MySqlHook
class BaseSensorOperator(BaseOperator):
@@ -1,5 +1,5 @@
import logging
from flux.models import BaseOperator
from airflow.models import BaseOperator
from subprocess import Popen, PIPE
@@ -1,4 +1,4 @@
from flux.models import BaseOperator
from airflow.models import BaseOperator
class DummyOperator(BaseOperator):
@@ -1,6 +1,6 @@
import logging
from flux.models import BaseOperator
from flux.hooks import MySqlHook
from airflow.models import BaseOperator
from airflow.hooks import MySqlHook
class MySqlOperator(BaseOperator):
+8 -8
View File
@@ -3,15 +3,15 @@ import sys
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
if 'FLUX_HOME' not in os.environ:
os.environ['FLUX_HOME'] = os.path.join(os.path.dirname(__file__), "..")
FLUX_HOME = os.environ['FLUX_HOME']
if 'AIRFLOW_HOME' not in os.environ:
os.environ['AIRFLOW_HOME'] = os.path.join(os.path.dirname(__file__), "..")
AIRFLOW_HOME = os.environ['AIRFLOW_HOME']
BASE_FOLDER = FLUX_HOME + '/flux'
BASE_FOLDER = AIRFLOW_HOME + '/airflow'
if BASE_FOLDER not in sys.path:
sys.path.append(BASE_FOLDER)
DAGS_FOLDER = FLUX_HOME + '/dags'
BASE_LOG_FOLDER = FLUX_HOME + "/logs"
DAGS_FOLDER = AIRFLOW_HOME + '/dags'
BASE_LOG_FOLDER = AIRFLOW_HOME + "/logs"
HIVE_HOME_PY = '/usr/lib/hive/lib/py'
RUN_AS_MASTER = True
JOB_HEARTBEAT_SEC = 5
@@ -19,8 +19,8 @@ ID_LEN = 250 # Used for dag_id and task_id VARCHAR length
LOG_FORMAT = \
'[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s'
Session = sessionmaker()
engine = create_engine('mysql://flux:flux@localhost/flux')
# engine = create_engine('sqlite:///' + BASE_FOLDER + '/flux.db' )
engine = create_engine('mysql://airflow:airflow@localhost/airflow')
# engine = create_engine('sqlite:///' + BASE_FOLDER + '/airflow.db' )
Session.configure(bind=engine)
HEADER = """\
_____ __
+1 -1
View File
@@ -35,7 +35,7 @@ def validate_key(k, max_length=250):
elif not re.match(r'^[A-Za-z0-9_]+$', k):
raise Exception(
"The key has to be made of alphanumeric characters and "
"undersfluxs exclusively")
"undersairflows exclusively")
else:
return True
File renamed without changes.
+13 -13
View File
@@ -17,17 +17,17 @@ from pygments.formatters import HtmlFormatter
import markdown
import chartkick
from flux.settings import Session
from flux import models
from flux.models import State
from flux import settings
from flux import utils
from airflow.settings import Session
from airflow import models
from airflow.models import State
from airflow import settings
from airflow import utils
dagbag = models.DagBag(settings.DAGS_FOLDER)
session = Session()
app = Flask(__name__)
app.secret_key = 'fluxified'
app.secret_key = 'airflowified'
# Init for chartkick, the python wrapper for highcharts
ck = Blueprint(
@@ -54,13 +54,13 @@ class HomeView(AdminIndexView):
@expose("/")
def index(self):
md = "".join(
open(settings.FLUX_HOME + '/README.md', 'r').readlines())
open(settings.AIRFLOW_HOME + '/README.md', 'r').readlines())
content = Markup(markdown.markdown(md))
return self.render('admin/index.html', content=content)
admin = Admin(app, name="Flux", index_view=HomeView(name='Home'))
admin = Admin(app, name="Airflow", index_view=HomeView(name='Home'))
class Flux(BaseView):
class Airflow(BaseView):
def is_visible(self):
return False
@@ -145,7 +145,7 @@ class Flux(BaseView):
downstream=downstream)
flash("{0} task instances have been cleared".format(count))
return redirect(url_for('flux.tree', dag_id=dag_id))
return redirect(url_for('airflow.tree', dag_id=dag_id))
dates = utils.date_range(
from_date, base_date, dag.schedule_interval)
@@ -369,7 +369,7 @@ class Flux(BaseView):
)
admin.add_view(Flux(name='DAGs'))
admin.add_view(Airflow(name='DAGs'))
# ------------------------------------------------
# Leveraging the admin for CRUD and browse on models
@@ -388,14 +388,14 @@ class ModelViewOnly(ModelView):
def filepath_formatter(view, context, model, name):
url = url_for('flux.code', dag_id=model.dag_id)
url = url_for('airflow.code', dag_id=model.dag_id)
short_fp = model.filepath.replace(settings.BASE_FOLDER + '/dags/', '')
link = Markup('<a href="{url}">{short_fp}</a>'.format(**locals()))
return link
def dag_formatter(view, context, model, name):
url = url_for('flux.tree', dag_id=model.dag_id, num_runs=45)
url = url_for('airflow.tree', dag_id=model.dag_id, num_runs=45)
link = Markup('<a href="{url}">{model.dag_id}</a>'.format(**locals()))
return link
File renamed without changes.
File renamed without changes.
File renamed without changes.
+1
View File
@@ -0,0 +1 @@
../../../docs/_build/html/
File renamed without changes.

Before

Width:  |  Height:  |  Size: 1.4 KiB

After

Width:  |  Height:  |  Size: 1.4 KiB

File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.

Before

Width:  |  Height:  |  Size: 30 KiB

After

Width:  |  Height:  |  Size: 30 KiB

File renamed without changes.

Before

Width:  |  Height:  |  Size: 37 KiB

After

Width:  |  Height:  |  Size: 37 KiB

File renamed without changes.

Before

Width:  |  Height:  |  Size: 34 KiB

After

Width:  |  Height:  |  Size: 34 KiB

File renamed without changes.
@@ -1,5 +1,5 @@
{% extends "admin/dag.html" %}
{% block title %}Flux - DAGs{% endblock %}
{% block title %}Airflow - DAGs{% endblock %}
{% block tail %}
{{ super() }}
@@ -1,5 +1,5 @@
{% extends "admin/dag.html" %}
{% block title %}Flux - DAGs{% endblock %}
{% block title %}Airflow - DAGs{% endblock %}
{% block body %}
{{ super() }}
@@ -1,6 +1,6 @@
{% extends "admin/master.html" %}
{% block title %}Flux - DAGs{% endblock %}
{% block title %}Airflow - DAGs{% endblock %}
{% block head_css %}
{{ super() }}
@@ -11,29 +11,29 @@
{% block body %}
<h2>DAG: {{ dag.dag_id }}</h2>
<ul class="nav nav-pills">
<li><a href="{{ url_for("flux.tree", dag_id=dag.dag_id, num_runs=45) }}">
<li><a href="{{ url_for("airflow.tree", dag_id=dag.dag_id, num_runs=45) }}">
<i class="icon-plus-sign"></i>Tree View
</a></li>
<li><a href="{{ url_for("flux.graph", dag_id=dag.dag_id) }}">
<li><a href="{{ url_for("airflow.graph", dag_id=dag.dag_id) }}">
<i class="icon-random"></i>
Graph View</a></li>
<li><a href="{{ url_for("flux.duration", dag_id=dag.dag_id) }}">
<li><a href="{{ url_for("airflow.duration", dag_id=dag.dag_id) }}">
<i class="icon-signal"></i>Task Duration
</a></li>
<li>
<a href="{{ url_for("flux.landing_times", dag_id=dag.dag_id) }}">
<a href="{{ url_for("airflow.landing_times", dag_id=dag.dag_id) }}">
<i class="icon-signal"></i>
Landing Times
</a>
</li>
<li>
<a href="{{ url_for("flux.gantt", dag_id=dag.dag_id) }}">
<a href="{{ url_for("airflow.gantt", dag_id=dag.dag_id) }}">
<i class="icon-align-left"></i>
Gantt
</a>
</li>
<li>
<a href="{{ url_for("flux.code", dag_id=dag.dag_id) }}">
<a href="{{ url_for("airflow.code", dag_id=dag.dag_id) }}">
<i class="icon-barcode"></i>
Code
</a>
@@ -115,7 +115,7 @@
}
$("#btn_log").click(function(){
url = "{{ url_for('flux.log') }}" +
url = "{{ url_for('airflow.log') }}" +
"?task_id=" + task_id +
"&dag_id=" + dag_id +
"&execution_date=" + execution_date;
@@ -123,7 +123,7 @@
});
$("#btn_clear").click(function(){
url = "{{ url_for('flux.tree') }}" +
url = "{{ url_for('airflow.tree') }}" +
"?action=clear" +
"&task_id=" + task_id +
"&dag_id=" + dag_id +
@@ -136,14 +136,14 @@
});
$("#btn_gantt").click(function(){
url = "{{ url_for('flux.gantt') }}" +
url = "{{ url_for('airflow.gantt') }}" +
"?dag_id=" + dag_id +
"&execution_date=" + execution_date;
window.location = url;
});
$("#btn_graph").click(function(){
url = "{{ url_for('flux.graph') }}" +
url = "{{ url_for('airflow.graph') }}" +
"?dag_id=" + dag_id +
"&execution_date=" + execution_date;
window.location = url;
File renamed without changes.
@@ -2,7 +2,7 @@
{% import 'admin/lib.html' as lib with context %}
{% import 'admin/static.html' as admin_static with context %}
{% block title %}Flux - DAGs{% endblock %}
{% block title %}Airflow - DAGs{% endblock %}
{% block head_css %}
{{ super() }}
File renamed without changes.
File renamed without changes.
@@ -1,5 +1,5 @@
{% extends "admin/dag.html" %}
{% block title %}Flux - DAGs{% endblock %}
{% block title %}Airflow - DAGs{% endblock %}
{% block head_css %}
{{ super() }}
+2 -2
View File
@@ -1,5 +1,5 @@
from flux.operators import BashOperator, MySqlOperator, DummyOperator
from flux.models import DAG
from airflow.operators import BashOperator, MySqlOperator, DummyOperator
from airflow.models import DAG
from datetime import datetime
default_args = {
+2 -2
View File
@@ -1,5 +1,5 @@
from flux.operators import BashOperator
from flux.models import DAG
from airflow.operators import BashOperator
from airflow.models import DAG
from datetime import datetime
default_args = {
+2 -2
View File
@@ -1,5 +1,5 @@
from flux.operators import BashOperator, MySqlOperator
from flux.models import DAG
from airflow.operators import BashOperator, MySqlOperator
from airflow.models import DAG
from datetime import datetime
default_args = {
+2 -2
View File
@@ -1,6 +1,6 @@
from flux import DAG
from airflow import DAG
from datetime import datetime, timedelta
from flux.operators import BashOperator
from airflow.operators import BashOperator
default_args = {
'owner': 'max',
+3 -3
View File
@@ -1,6 +1,6 @@
from flux.operators import MySqlOperator
from flux.executors import SequentialExecutor
from flux import DAG
from airflow.operators import MySqlOperator
from airflow.executors import SequentialExecutor
from airflow import DAG
from datetime import datetime
# Setting some default operator parameters
+2 -2
View File
@@ -1,5 +1,5 @@
from flux.operators import BashOperator
from flux.models import DAG
from airflow.operators import BashOperator
from airflow.models import DAG
from datetime import datetime, timedelta
default_args = {
+4 -4
View File
@@ -1,4 +1,4 @@
import flux as flux
import airflow as airflow
from datetime import datetime
default_args = {
@@ -7,13 +7,13 @@ default_args = {
'mysql_dbid': 'local_mysql',
}
dag = flux.DAG('test_mysql')
dag = airflow.DAG('test_mysql')
create = flux.operators.MySqlOperator(task_id='create',
create = airflow.operators.MySqlOperator(task_id='create',
sql='CREATE TABLE IF NOT EXISTS tmp (tmp INT);', **default_args)
dag.add_task(create)
ms = flux.operators.MySqlSensorOperator(task_id='sensor',
ms = airflow.operators.MySqlSensorOperator(task_id='sensor',
sql='SELECT COUNT(*) FROM tmp;', **default_args)
dag.add_task(ms)
ms.set_upstream(create)
+2 -2
View File
@@ -1,5 +1,5 @@
from flux.operators import ExternalTaskSensor
from flux import DAG
from airflow.operators import ExternalTaskSensor
from airflow import DAG
from datetime import datetime
# Setting some default operator parameters
+9 -9
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
# Flux documentation build configuration file, created by
# Airflow documentation build configuration file, created by
# sphinx-quickstart on Thu Oct 9 20:50:01 2014.
#
# This file is execfile()d with the current directory set to its
@@ -14,12 +14,12 @@
import sys
import os
from flux import settings
from airflow import settings
# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
sys.path.insert(0, settings.FLUX_HOME + "/src/flux")
sys.path.insert(0, settings.AIRFLOW_HOME + "/src/airflow")
# -- General configuration ------------------------------------------------
@@ -49,7 +49,7 @@ source_suffix = '.rst'
master_doc = 'index'
# General information about the project.
project = u'Flux'
project = u'Airflow'
copyright = u'2014, Maxime Beauchemin'
# The version info for the project you're documenting, acts as replacement for
@@ -184,7 +184,7 @@ html_use_index = True
#html_file_suffix = None
# Output file base name for HTML help builder.
htmlhelp_basename = 'Fluxdoc'
htmlhelp_basename = 'Airflowdoc'
# -- Options for LaTeX output ---------------------------------------------
@@ -204,7 +204,7 @@ latex_elements = {
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
('index', 'Flux.tex', u'Flux Documentation',
('index', 'Airflow.tex', u'Airflow Documentation',
u'Maxime Beauchemin', 'manual'),
]
@@ -234,7 +234,7 @@ latex_documents = [
# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [
('index', 'flux', u'Flux Documentation',
('index', 'airflow', u'Airflow Documentation',
[u'Maxime Beauchemin'], 1)
]
@@ -248,8 +248,8 @@ man_pages = [
# (source start file, target name, title, author,
# dir menu entry, description, category)
texinfo_documents = [
('index', 'Flux', u'Flux Documentation',
u'Maxime Beauchemin', 'Flux', 'One line description of project.',
('index', 'Airflow', u'Airflow Documentation',
u'Maxime Beauchemin', 'Airflow', 'One line description of project.',
'Miscellaneous'),
]
+5 -5
View File
@@ -1,15 +1,15 @@
Flux's Documentation
Airflow's Documentation
================================
Flux is a system to programmaticaly author, schedule and monitor data pipelines.
Airflow is a system to programmaticaly author, schedule and monitor data pipelines.
Use the Flux library to define workflows as directed acyclic graphs (DAGs) of data related tasks. Command line utilities make it easy to run parts of workflows interactively, and commiting pipelines into production is all it takes for the master scheduler to run the pipelines with the schedule and dependencies specified.
Use the Airflow library to define workflows as directed acyclic graphs (DAGs) of data related tasks. Command line utilities make it easy to run parts of workflows interactively, and commiting pipelines into production is all it takes for the master scheduler to run the pipelines with the schedule and dependencies specified.
The Flux UI make it easy to visualize pipelines running in production, monitor progress and pinpoint and troubleshoot issues when needed.
The Airflow UI make it easy to visualize pipelines running in production, monitor progress and pinpoint and troubleshoot issues when needed.
Principles
----------
- **Dynamic:** Flux has intrinsec support for dynamic pipeline
- **Dynamic:** Airflow has intrinsec support for dynamic pipeline
generation: you can write pipelines, as well as writing code that
defines pipeline.
- **Interactivity:** the libraries are intuitive so that writting /
+10 -10
View File
@@ -32,8 +32,8 @@ Required environment variable, add this to your .bashrc or .bash_profile
::
export FLUX_HOME=~/Flux
export PATH=$PATH:$FLUX_HOME/flux/bin
export AIRFLOW_HOME=~/Airflow
export PATH=$PATH:$AIRFLOW_HOME/airflow/bin
# also run it, or source it, or start a new shell
source ~/.bashrc
@@ -45,8 +45,8 @@ Create a python virtualenv
virtualenv env # creates the environment
source init.sh # activates the environment
Use pip to install the python packages required by Flux
'''''''''''''''''''''''''''''''''''''''''''''''''''''''
Use pip to install the python packages required by Airflow
''''''''''''''''''''''''''''''''''''''''''''''''''''''''''
::
@@ -61,19 +61,19 @@ fine.
::
# If :
$ mysql -u root -p
mysql>
CREATE DATABASE flux;
CREATE USER 'flux'@'localhost' IDENTIFIED BY 'flux';
GRANT ALL PRIVILEGES ON flux.* TO 'flux'@'localhost';
CREATE DATABASE airflow;
CREATE USER 'airflow'@'localhost' IDENTIFIED BY 'airflow';
GRANT ALL PRIVILEGES ON airflow.* TO 'airflow'@'localhost';
Get things started
''''''''''''''''''''
::
# Creating the necessary tables in the database
flux initdb
airflow initdb
# Start the web server!
flux webserver --port 8080
airflow webserver --port 8080
+44 -44
View File
@@ -1,5 +1,5 @@
Flux tutorial
Airflow tutorial
=============
This tutorial walks through some of the fundamental concepts, objects
@@ -8,15 +8,15 @@ and their usage.
Hooks
-----
Hooks are a simple abstraction layer on systems Flux interacts with. You
Hooks are a simple abstraction layer on systems Airflow interacts with. You
should expect a lot more consistency across hooks than you would with
the different systems' underlying libraries. You should also expect a
higher level of abstraction.
Connection information is stored in the Flux metadata database, so that
Connection information is stored in the Airflow metadata database, so that
you don't need to hard code or remember this connection information. In
the bellow example, we connect to a MySQL database by specifying the
mysql\_dbid, which looks up Flux's metadata to get the actual hostname,
mysql\_dbid, which looks up Airflow's metadata to get the actual hostname,
login, password, and schema name behind the scene.
Common methods: \* Get a recordset \* Extract a csv file \* Run a
@@ -31,12 +31,12 @@ statement \* Load into a table from a csv file \* Get a pandas dataframe
logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG, datefmt='%I:%M:%S')
.. code:: python
from flux.hooks import MySqlHook
from airflow.hooks import MySqlHook
mysql_hook = MySqlHook(mysql_dbid='local_mysql')
sql = """
SELECT table_schema, table_name
FROM information_schema.tables
WHERE table_schema = 'flux'
WHERE table_schema = 'airflow'
"""
mysql_hook.get_records(sql)
@@ -44,17 +44,17 @@ statement \* Load into a table from a csv file \* Get a pandas dataframe
.. parsed-literal::
(('flux', 'dag'),
('flux', 'dag_pickle'),
('flux', 'dag_picle'),
('flux', 'db_connection'),
('flux', 'deleteme'),
('flux', 'job'),
('flux', 'log'),
('flux', 'task'),
('flux', 'task_instance'),
('flux', 'tmp'),
('flux', 'user'))
(('airflow', 'dag'),
('airflow', 'dag_pickle'),
('airflow', 'dag_picle'),
('airflow', 'db_connection'),
('airflow', 'deleteme'),
('airflow', 'job'),
('airflow', 'log'),
('airflow', 'task'),
('airflow', 'task_instance'),
('airflow', 'tmp'),
('airflow', 'user'))
@@ -77,12 +77,12 @@ instances.
Bellow we run a simple remote MySQL statement, over a date range. The
task.run() method will instanciate many task runs for the schedule
specified, and run them, storing the state in the Flux database. If you
specified, and run them, storing the state in the Airflow database. If you
were to re-run this, it would say it already succeded.
.. code:: python
from flux.operators import MySqlOperator
from airflow.operators import MySqlOperator
from datetime import datetime, timedelta
sql = """
@@ -128,8 +128,8 @@ and their associated task instance run states.
.. code:: python
from flux.operators import MySqlOperator
from flux import DAG
from airflow.operators import MySqlOperator
from airflow import DAG
from datetime import datetime
# Setting some default operator parameters
@@ -166,8 +166,8 @@ and their associated task instance run states.
.. parsed-literal::
INFO:Adding to queue: ./flux run test_dag mysql_fisrt 2014-09-01T00:00:00 --pickle 10
INFO:Adding to queue: ./flux run test_dag mysql_second 2014-09-01T00:00:00 --pickle 10
INFO:Adding to queue: ./airflow run test_dag mysql_fisrt 2014-09-01T00:00:00 --pickle 10
INFO:Adding to queue: ./airflow run test_dag mysql_second 2014-09-01T00:00:00 --pickle 10
INFO:Run summary:
@@ -228,22 +228,22 @@ false negative, rerun subsection DAGs.
%%bash
# Printing the --help for the main and subcommands
cd /home/mistercrunch/Flux
./flux --help
cd /home/mistercrunch/Airflow
./airflow --help
echo ============================================================================
./flux backfill -h
./airflow backfill -h
echo ============================================================================
./flux clear -h
./airflow clear -h
echo ============================================================================
./flux run -h
./airflow run -h
echo ============================================================================
./flux webserver -h
./airflow webserver -h
echo ============================================================================
./flux master -h
./airflow master -h
.. parsed-literal::
usage: flux_bin.py [-h] {backfill,clear,run,webserver,master} ...
usage: airflow_bin.py [-h] {backfill,clear,run,webserver,master} ...
positional arguments:
{backfill,clear,run,webserver,master}
@@ -251,13 +251,13 @@ false negative, rerun subsection DAGs.
backfill Run subsections of a DAG for a specified date range
clear Clear a set of task instance, as if they never ran
run Run a single task instance
webserver Start a Flux webserver instance
webserver Start a Airflow webserver instance
master Start a master scheduler instance
optional arguments:
-h, --help show this help message and exit
============================================================================
usage: flux_bin.py backfill [-h] [-t TASK_REGEX] [-s START_DATE] [-e END_DATE] [-m] [-sd SUBDIR] dag_id
usage: airflow_bin.py backfill [-h] [-t TASK_REGEX] [-s START_DATE] [-e END_DATE] [-m] [-sd SUBDIR] dag_id
positional arguments:
dag_id The id of the dag to run
@@ -274,7 +274,7 @@ false negative, rerun subsection DAGs.
-sd SUBDIR, --subdir SUBDIR
File location or directory from which to look for the dag
============================================================================
usage: flux_bin.py clear [-h] [-t TASK_REGEX] [-s START_DATE] [-e END_DATE] [-sd SUBDIR] dag_id
usage: airflow_bin.py clear [-h] [-t TASK_REGEX] [-s START_DATE] [-e END_DATE] [-sd SUBDIR] dag_id
positional arguments:
dag_id The id of the dag to run
@@ -290,7 +290,7 @@ false negative, rerun subsection DAGs.
-sd SUBDIR, --subdir SUBDIR
File location or directory from which to look for the dag
============================================================================
usage: flux_bin.py run [-h] [-sd SUBDIR] [-m] [-f] [-i] [-p PICKLE] dag_id task_id execution_date
usage: airflow_bin.py run [-h] [-sd SUBDIR] [-m] [-f] [-i] [-p PICKLE] dag_id task_id execution_date
positional arguments:
dag_id The id of the dag to run
@@ -308,13 +308,13 @@ false negative, rerun subsection DAGs.
-p PICKLE, --pickle PICKLE
Serialized pickle object of the entire dag (used internally)
============================================================================
usage: flux_bin.py webserver [-h] [-p PORT]
usage: airflow_bin.py webserver [-h] [-p PORT]
optional arguments:
-h, --help show this help message and exit
-p PORT, --port PORT Set the port on which to run the web server
============================================================================
usage: flux_bin.py master [-h] [-d DAG_ID] [-sd SUBDIR]
usage: airflow_bin.py master [-h] [-d DAG_ID] [-sd SUBDIR]
optional arguments:
-h, --help show this help message and exit
@@ -328,17 +328,17 @@ false negative, rerun subsection DAGs.
%%bash
# Example run command
cd $FLUX_HOME
./flux run example_2 runme_1 2014-09-01 -sd /home/mistercrunch/Flux/dags/examples/example2.py
cd $AIRFLOW_HOME
./airflow run example_2 runme_1 2014-09-01 -sd /home/mistercrunch/Airflow/dags/examples/example2.py
# Printing the log
cat /home/mistercrunch/Flux/logs/example_2/runme_1/2014-09-01T00:00:00
cat /home/mistercrunch/Airflow/logs/example_2/runme_1/2014-09-01T00:00:00
.. parsed-literal::
Logging into: /home/mistercrunch/Flux/logs/example_2/runme_1/2014-09-01T00:00:00
[2014-10-08 15:11:39,177] {models.py:44} INFO - Filling up the DagBag from /home/mistercrunch/Flux/dags/examples/example2.py
[2014-10-08 15:11:39,177] {models.py:53} INFO - Importing /home/mistercrunch/Flux/dags/examples/example2.py
Logging into: /home/mistercrunch/Airflow/logs/example_2/runme_1/2014-09-01T00:00:00
[2014-10-08 15:11:39,177] {models.py:44} INFO - Filling up the DagBag from /home/mistercrunch/Airflow/dags/examples/example2.py
[2014-10-08 15:11:39,177] {models.py:53} INFO - Importing /home/mistercrunch/Airflow/dags/examples/example2.py
[2014-10-08 15:11:39,272] {models.py:325} INFO -
--------------------------------------------------------------------------------
New run starting @2014-10-08T15:11:39.272355
@@ -380,7 +380,7 @@ Here's the source for the MySqlOperator
Executors
---------
Executors are an abrastraction on top of systems that can run Flux task
Executors are an abrastraction on top of systems that can run Airflow task
instances. The default LocalExecutor is a simple implementation of
Python's multiprocessing with a simple joinable queue.
+1 -1
View File
@@ -1,6 +1,6 @@
UI / Screenshots
=================
The Flux UI make it easy to monitor and troubleshoot your data pipelines.
The Airflow UI make it easy to monitor and troubleshoot your data pipelines.
Tree View
-1
View File
@@ -1 +0,0 @@
/home/mistercrunch/Flux/docs/_build/html/
+2 -2
View File
@@ -1,3 +1,3 @@
#!/bin/bash
source $FLUX_HOME/env/bin/activate
export PYTHONPATH=$FLUX_HOME
source $AIRFLOW_HOME/env/bin/activate
export PYTHONPATH=$AIRFLOW_HOME
+8 -8
View File
@@ -7,18 +7,18 @@ reqs = (str(ir.req) for ir in parse_requirements('requirements.txt'))
print reqs
setup(
name='flux',
name='airflow',
version='0.1',
packages=[
'flux',
'flux.operators',
'flux.executors',
'flux.hooks',
'flux.www',
'airflow',
'airflow.operators',
'airflow.executors',
'airflow.hooks',
'airflow.www',
],
scripts=['flux/bin/flux'],
scripts=['airflow/bin/airflow'],
install_requires=reqs,
author='Maxime Beauchemin',
author_email='maximebeauchemin@gmail.com',
url='https://github.com/mistercrunch/Flux',
url='https://github.com/mistercrunch/Airflow',
)