airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bo...@apache.org
Subject incubator-airflow git commit: [AIRFLOW-216] Add Sqoop Hook and Operator
Date Tue, 14 Jun 2016 11:15:35 GMT
Repository: incubator-airflow
Updated Branches:
  refs/heads/master 8d501b0ce -> 717a4aeee


[AIRFLOW-216] Add Sqoop Hook and Operator

This patch adds a Sqoop hook and operator that implements Sqoop import.
The hook is a wrapper around the sqoop 1 binary.

* Closes #1576 from jwi078/sqoop_operator


Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/717a4aee
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/717a4aee
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/717a4aee

Branch: refs/heads/master
Commit: 717a4aeee3bd0883067c82a3174ad8e2b7dbe8ec
Parents: 8d501b0
Author: JohanW <johan078@gmail.com>
Authored: Tue Jun 14 13:14:55 2016 +0200
Committer: Bolke de Bruin <bolke@xs4all.nl>
Committed: Tue Jun 14 13:14:59 2016 +0200

----------------------------------------------------------------------
 airflow/contrib/hooks/sqoop_hook.py         | 223 +++++++++++++++++++++++
 airflow/contrib/operators/sqoop_operator.py |  91 +++++++++
 2 files changed, 314 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/717a4aee/airflow/contrib/hooks/sqoop_hook.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/hooks/sqoop_hook.py b/airflow/contrib/hooks/sqoop_hook.py
new file mode 100644
index 0000000..f213fc4
--- /dev/null
+++ b/airflow/contrib/hooks/sqoop_hook.py
@@ -0,0 +1,223 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+This module contains a sqoop 1 hook
+"""
+
+from airflow.hooks.base_hook import BaseHook
+from airflow.exceptions import AirflowException
+
+import logging
+import subprocess
+
+log = logging.getLogger(__name__)
+
+
+class SqoopHook(BaseHook):
+    """
+    This Hook is a wrapper around the sqoop 1 binary. To be able to use te hook
+    it is required that "sqoop" is in the PATH.
+    :param hive_home: (from json) The location of hive-site.xml
+    :type hive_home: str
+    :param job_tracker: (from json) <local|jobtracker:port> specify a job tracker
+    :type job_tracker: str
+    :param namenode: (from json) specify a namenode
+    :type namenode: str
+    :param lib_jars: (from json) specify comma separated jar files to
+        include in the classpath.
+    :type lib_jars: str
+    :param files: (from json) specify comma separated files to be copied
+        to the map reduce cluster
+    :type files: (from json) str
+    :param archives: (from json)  specify comma separated archives to be
+        unarchived on the compute machines.
+    :type archives: str
+    """
+    def __init__(self, conn_id='sqoop_default'):
+        conn = self.get_connection(conn_id)
+        self.hive_home = conn.extra_dejson.get('hive_home', None)
+        self.job_tracker = conn.extra_dejson.get('job_tracker', None)
+        self.namenode = conn.extra_dejson.get('namenode', None)
+        self.lib_jars = conn.extra_dejson.get('libjars', None)
+        self.files = conn.extra_dejson.get('files', None)
+        self.archives = conn.extra_dejson.get('archives', None)
+        self.conn = conn
+
+    def get_conn(self):
+        pass
+
+    def Popen(self, cmd, export=False, **kwargs):
+        """
+        Remote Popen
+
+        :param cmd: command to remotely execute
+        :param kwargs: extra arguments to Popen (see subprocess.Popen)
+        :return: handle to subprocess
+        """
+        prefixed_cmd = self._prepare_command(cmd, export=export)
+        print prefixed_cmd
+        return subprocess.Popen(prefixed_cmd, **kwargs)
+
+    def _prepare_command(self, cmd, export=False):
+
+        connection_cmd = ""
+
+        if export:
+            connection_cmd = ["sqoop", "export", "--verbose"]
+        else:
+            connection_cmd = ["sqoop", "import", "--verbose"]
+
+        if self.job_tracker:
+            connection_cmd += ["-jt", self.job_tracker]
+        if self.conn.login:
+            connection_cmd += ["--username", self.conn.login]
+        # todo: put this in a password file
+        if self.conn.password:
+            connection_cmd += ["--password", self.conn.password]
+        if self.lib_jars:
+            connection_cmd += ["-libjars", self.lib_jars]
+        if self.files:
+            connection_cmd += ["-files", self.files]
+        if self.namenode:
+            connection_cmd += ["-fs", self.namenode]
+        if self.archives:
+            connection_cmd += ["-archives", self.archives]
+
+        connection_cmd += ["--connect", "{}:{}/{}".format(self.conn.host, self.conn.port,
self.conn.schema)]
+        connection_cmd += cmd
+
+        return connection_cmd
+
+    def _import_cmd(self, target_dir,
+                    append=False, type="text",
+                    num_mappers=None, split_by=None):
+
+        cmd = ["--target-dir", target_dir]
+
+        if not num_mappers:
+            num_mappers = 1
+
+        cmd += ["--num-mappers", str(num_mappers)]
+
+        if split_by:
+            cmd += ["--split-by", split_by]
+
+        if append:
+            cmd += ["--append"]
+
+        if type == "avro":
+            cmd += ["--as-avrodatafile"]
+        elif type == "sequence":
+            cmd += ["--as-sequencefile"]
+        else:
+            cmd += ["--as-textfile"]
+
+        return cmd
+
+    def import_table(self, table, target_dir,
+                     append=False, type="text", columns=None,
+                     num_mappers=None, split_by=None, where=None):
+        """
+        Imports table from remote location to target dir. Arguments are
+        copies of direct sqoop command line arguments
+        :param table: Table to read
+        :param target_dir: HDFS destination dir
+        :param append: Append data to an existing dataset in HDFS
+        :param type: "avro", "sequence", "text" Imports data to into the specified
+            format. Defaults to text.
+        :param columns: <col,col,col…> Columns to import from table
+        :param num_mappers: Use n map tasks to import in parallel
+        :param split_by: Column of the table used to split work units
+        :param where: WHERE clause to use during import
+        """
+        cmd = self._import_cmd(target_dir, append, type,
+                               num_mappers, split_by)
+        cmd += ["--table", table]
+        if columns:
+            cmd += ["--columns", columns]
+        if where:
+            cmd += ["--where", where]
+
+        p = self.Popen(cmd, export=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        output, stderr = p.communicate()
+
+        if p.returncode != 0:
+            # I like this better: RemoteCalledProcessError(p.returncode, cmd, self.host,
output=output)
+            raise AirflowException("Cannot execute {} on {}. Error code is: "
+                                   "{}. Output: {}, Stderr: {}"
+                                   .format(cmd, self.conn.host,
+                                           p.returncode, output, stderr))
+
+    def _export_cmd(self, export_dir, num_mappers=None):
+
+        cmd = ["--export-dir", export_dir]
+
+        if not num_mappers:
+            num_mappers = 1
+
+        cmd += ["--num-mappers", str(num_mappers)]
+
+        return cmd
+
+    def export_table(self, table, export_dir,
+                     num_mappers=None):
+        """
+        Exports Hive table to remote location. Arguments are copies of direct
+        sqoop command line Arguments
+        :param table: Table remote destination
+        :param export_dir: Hive table to export
+        :param num_mappers: Use n map tasks to import in parallel
+        """
+
+        cmd = self._export_cmd(export_dir, num_mappers)
+        cmd += ["--table", table]
+
+        p = self.Popen(cmd, export=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        output, stderr = p.communicate()
+
+        if p.returncode != 0:
+            # I like this better: RemoteCalledProcessError(p.returncode, cmd, self.host,
output=output)
+            raise AirflowException("Cannot execute {} on {}. Error code is: "
+                                   "{}. Output: {}, Stderr: {}"
+                                   .format(cmd, self.conn.host,
+                                           p.returncode, output, stderr))
+
+    def import_query(self, query, target_dir,
+                     append=False, type="text",
+                     num_mappers=None, split_by=None):
+        """
+
+        :param query: Free format query to run
+        :param target_dir: HDFS destination dir
+        :param append: Append data to an existing dataset in HDFS
+        :param type: "avro", "sequence", "text" Imports data to into the specified
+            format. Defaults to text.
+        :param num_mappers: Use n map tasks to import in parallel
+        :param split_by: Column of the table used to split work units
+        """
+        cmd = self._import_cmd(target_dir, append, type,
+                               num_mappers, split_by)
+        cmd += ["--query", query]
+
+        p = self.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        output, stderr = p.communicate()
+
+        if p.returncode != 0:
+            # I like this better: RemoteCalledProcessError(p.returncode, cmd, self.host,
output=output)
+            raise AirflowException("Cannot execute {} on {}. Error code is: "
+                                   "{}. Output: {}, Stderr: {}"
+                                   .format(cmd, self.conn.host,
+                                           p.returncode, output, stderr))

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/717a4aee/airflow/contrib/operators/sqoop_operator.py
----------------------------------------------------------------------
diff --git a/airflow/contrib/operators/sqoop_operator.py b/airflow/contrib/operators/sqoop_operator.py
new file mode 100644
index 0000000..8bf1c05
--- /dev/null
+++ b/airflow/contrib/operators/sqoop_operator.py
@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+This module contains a sqoop 1 operator
+"""
+
+from airflow.models import BaseOperator
+from airflow.utils.decorators import apply_defaults
+from airflow.contrib.hooks.sqoop_hook import SqoopHook
+
+
+class SqoopOperator(BaseOperator):
+    """
+    execute sqoop job
+    """
+    @apply_defaults
+    def __init__(self,
+                 conn_id='sqoop_default',
+                 type_cmd='import',
+                 table='',
+                 target_dir=None,
+                 append=None,
+                 type=None,
+                 columns=None,
+                 num_mappers='1',
+                 split_by=None,
+                 where=None,
+                 export_dir=None,
+                 *args,
+                 **kwargs):
+        """
+        :param conn_id: str
+	:param type_cmd: str specify command to execute "export" or "import"
+        :param table: Table to read
+        :param target_dir: HDFS destination dir
+        :param append: Append data to an existing dataset in HDFS
+        :param type: "avro", "sequence", "text" Imports data to into the specified
+           format. Defaults to text.
+        :param columns: <col,col,col> Columns to import from table
+        :param num_mappers: U n map task to import/export in parallel
+        :param split_by: Column of the table used to split work units
+        :param where: WHERE clause to use during import
+        :param export_dir: HDFS Hive database directory to export
+        """
+        super(SqoopOperator, self).__init__(*args, **kwargs)
+        self.conn_id = conn_id
+        self.type_cmd = type_cmd
+        self.table = table
+        self.target_dir = target_dir
+        self.append = append
+        self.type = type
+        self.columns = columns
+        self.num_mappers = num_mappers
+        self.split_by = split_by
+        self.where = where
+        self.export_dir = export_dir
+
+    def execute(self, context):
+        """
+        Execute sqoop job
+        """
+        hook = SqoopHook(conn_id=self.conn_id)
+
+        if self.type_cmd is 'export':
+            hook.export_table(
+                table=self.table,
+                export_dir=self.export_dir,
+                num_mappers=self.num_mappers)
+        else:
+            hook.import_table(
+                table=self.table,
+                target_dir=self.target_dir,
+                append=self.append,
+                type=self.type,
+                columns=self.columns,
+                num_mappers=self.num_mappers,
+                split_by=self.split_by,
+                where=self.where)


Mime
View raw message