diff --git a/.gitignore b/.gitignore index 81387a5b7fb..75a9ca8ee08 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +airflow.db env logs build diff --git a/airflow/executors/__init__.py b/airflow/executors/__init__.py index 07a0af35574..1b310fcb37f 100644 --- a/airflow/executors/__init__.py +++ b/airflow/executors/__init__.py @@ -1,4 +1,4 @@ from base_executor import LocalExecutor from base_executor import SequentialExecutor -DEFAULT_EXECUTOR = LocalExecutor() +DEFAULT_EXECUTOR = SequentialExecutor() diff --git a/airflow/hooks/__init__.py b/airflow/hooks/__init__.py index 1b3f5ebfbbb..5296c6e30d1 100644 --- a/airflow/hooks/__init__.py +++ b/airflow/hooks/__init__.py @@ -1 +1,2 @@ from mysql_hook import MySqlHook +from hive_hook import HiveHook diff --git a/airflow/hooks/hive/__init__.py b/airflow/hooks/hive/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000 diff --git a/airflow/hooks/hive/hive_hook.py b/airflow/hooks/hive_hook.py similarity index 90% rename from airflow/hooks/hive/hive_hook.py rename to airflow/hooks/hive_hook.py index 7bf6d9bb3dc..dfae81b3045 100644 --- a/airflow/hooks/hive/hive_hook.py +++ b/airflow/hooks/hive_hook.py @@ -1,4 +1,5 @@ import logging +import subprocess import sys import os from airflow import settings @@ -16,7 +17,7 @@ from hive_service import ThriftHive from airflow.hooks.base_hook import BaseHook METASTORE_THRIFT_HOST = "localhost" -METASTORE_THRIFT_PORT = 10000 +METASTORE_THRIFT_PORT = 8083 class HiveHook(BaseHook): @@ -66,6 +67,12 @@ class HiveHook(BaseHook): self.hive.execute(hql) self.transport.close() + def run_cli(self, hql, schema=None): + if schema: + hql = "USE {schema};\n{hql}".format(**locals()) + sp = subprocess.Popen(['hive', '-e', hql]) + sp.wait() + if __name__ == "__main__": hh = HiveHook() hql = "SELECT * FROM fct_nights_booked WHERE ds='2014-10-01' LIMIT 2" diff --git a/airflow/operators/__init__.py b/airflow/operators/__init__.py index ee57264980b..6c2345efd7d 100644 --- a/airflow/operators/__init__.py +++ b/airflow/operators/__init__.py @@ -1,5 +1,6 @@ from bash_operator import BashOperator from mysql_operator import MySqlOperator +from hive_operator import HiveOperator from base_sensor_operator import MySqlSensorOperator from base_sensor_operator import ExternalTaskSensor from dummy_operator import DummyOperator diff --git a/airflow/operators/hive_operator.py b/airflow/operators/hive_operator.py new file mode 100644 index 00000000000..ab1cf2a5d3f --- /dev/null +++ b/airflow/operators/hive_operator.py @@ -0,0 +1,31 @@ +import logging +from airflow.models import BaseOperator +from airflow.hooks import HiveHook + + +class HiveOperator(BaseOperator): + """ + Executes sql code in a specific mysql database. + """ + + __mapper_args__ = { + 'polymorphic_identity': 'HiveOperator' + } + template_fields = ('hql',) + + def __init__(self, hql, hive_dbid, *args, **kwargs): + """ + Parameters: + mysql_dbid: reference to a specific mysql database + sql: the sql code you to be executed + """ + super(HiveOperator, self).__init__(*args, **kwargs) + + self.hive_dbid = hive_dbid + self.hook = HiveHook(hive_dbid=hive_dbid) + self.hql = hql + + + def execute(self, execution_date): + logging.info('Executing: ' + self.hql) + self.hook.run_cli(hql=self.hql, schema=self.hive_dbid) diff --git a/airflow/settings.py b/airflow/settings.py index 12fb9dbb336..9b2a0742ac0 100644 --- a/airflow/settings.py +++ b/airflow/settings.py @@ -21,8 +21,8 @@ LOG_FORMAT = \ '[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s' Session = sessionmaker() -engine = create_engine('mysql://airflow:airflow@localhost/airflow') -# engine = create_engine('sqlite:///' + BASE_FOLDER + '/airflow.db' ) +#engine = create_engine('mysql://airflow:airflow@localhost/airflow') +engine = create_engine('sqlite:///' + BASE_FOLDER + '/airflow.db' ) Session.configure(bind=engine) HEADER = """\ .__ _____.__ diff --git a/setup.py b/setup.py index ecedd3bc96f..fe04bef17da 100644 --- a/setup.py +++ b/setup.py @@ -4,14 +4,13 @@ from pip.req import parse_requirements # Reading the requirements from pip's requirements.txt reqs = (str(ir.req) for ir in parse_requirements('requirements.txt')) -print reqs setup( name='airflow', version='0.1', packages=[ - 'airflow', - 'airflow.operators', + 'airflow', + 'airflow.operators', 'airflow.executors', 'airflow.hooks', 'airflow.www',