Adding HiveOperator

.gitignore

@@ -1,3 +1,4 @@
+airflow.db
 env
 logs
 build

airflow/executors/__init__.py

@@ -1,4 +1,4 @@
 from base_executor import LocalExecutor
 from base_executor import SequentialExecutor
 
-DEFAULT_EXECUTOR = LocalExecutor()
+DEFAULT_EXECUTOR = SequentialExecutor()

airflow/hooks/__init__.py

@@ -1 +1,2 @@
 from mysql_hook import MySqlHook
+from hive_hook import HiveHook

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"

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

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)

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 = """\
        .__         _____.__                 

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',