瀏覽代碼

HUE-6659 [import] Support RDBMs as input via Sqoop

Prachi Poddar 8 年之前
父節點
當前提交
46e2e39

+ 1 - 1
apps/oozie/src/oozie/models2.py

@@ -3931,7 +3931,7 @@ class WorkflowBuilder():
         'name': 'sqoop-%s' % node_id[:4],
         "type": "sqoop-document-widget",
         "properties":{
-              "command": "",
+              "statement": "",
               "arguments": [],
               "retry_max": [],
               "retry_interval": [],

+ 21 - 2
desktop/libs/indexer/src/indexer/api3.py

@@ -25,7 +25,9 @@ from desktop.lib import django_mako
 from desktop.lib.django_util import JsonResponse
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.models import Document2
+from librdbms.server import dbms as rdbms
 from notebook.connectors.base import get_api, Notebook
+from notebook.connectors.rdbms import Assist
 from notebook.decorators import api_error_handler
 from notebook.models import make_notebook
 
@@ -99,6 +101,8 @@ def guess_format(request):
       raise PopupException('Hive table format %s is not supported.' % table_metadata.details['properties']['format'])
   elif file_format['inputFormat'] == 'query':
     format_ = {"quoteChar": "\"", "recordSeparator": "\\n", "type": "csv", "hasHeader": False, "fieldSeparator": "\u0001"}
+  elif file_format['inputFormat'] == 'rdbms':
+    format_ = {"type": "csv"}
 
   format_['status'] = 0
   return JsonResponse(format_)
@@ -145,6 +149,20 @@ def guess_field_types(request):
             for col in sample.meta
         ]
     }
+  elif file_format['inputFormat'] == 'rdbms':
+    query_server = rdbms.get_query_server_config(server=file_format['rdbmsName'])
+    db = rdbms.get(request.user, query_server=query_server)
+    assist = Assist(db)
+    sample = assist.get_sample_data(database=file_format['rdbmsDatabaseName'], table=file_format['rdbmsTableName'])
+    table_metadata = db.get_columns(file_format['rdbmsDatabaseName'], file_format['rdbmsTableName'], names_only=False)
+
+    format_ = {
+        "sample": list(sample.rows()),
+        "columns": [
+            Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
+            for col in table_metadata
+        ]
+    }
 
   return JsonResponse(format_)
 
@@ -169,9 +187,11 @@ def importer_submit(request):
       job_handle = _create_index(request.user, request.fs, client, source, destination, index_name)
   elif destination['ouputFormat'] == 'database':
     job_handle = _create_database(request, source, destination, start_time)
+  elif destination['outputFormat'] == 'file' and source['inputFormat'] == 'rdbms':
+    job_handle = run_morphline(request, source, 'sqoop')
   else:
     job_handle = _create_table(request, source, destination, start_time)
-
+  print JsonResponse(job_handle)
   return JsonResponse(job_handle)
 
 
@@ -277,7 +297,6 @@ def _index(request, file_format, collection_name, query=None, start_time=None, l
       fields=request.POST.get('fields', schema_fields),
       unique_key_field=unique_field
     )
-
   if file_format['inputFormat'] == 'table':
     db = dbms.get(request.user)
     table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])

+ 7 - 1
desktop/libs/indexer/src/indexer/conf.py

@@ -86,8 +86,14 @@ CONFIG_INDEXER_LIBS_PATH = Config(
   help=_t("oozie workspace template for indexing:"),
   type=str,
   default='/tmp/smart_indexer_lib'
-  )
+)
 
+ENABLE_SQOOP = Config(
+  key="enable_sqoop",
+  help=_t("Flag to turn on Sqoop imports."),
+  type=bool,
+  default=False
+)
 
 # Unused
 BATCH_INDEXER_PATH = Config(

+ 69 - 0
desktop/libs/indexer/src/indexer/rdbms_indexer.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you 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.import logging
+
+import logging
+
+from librdbms.server import dbms
+from notebook.connectors.rdbms import Assist
+
+
+LOG = logging.getLogger(__name__)
+
+
+class RdbmsIndexer():
+
+  def __init__(self, user, db_conf_name):
+    self.user = user
+    self.db_conf_name = db_conf_name
+
+  def guess_type(self):
+    return {}
+
+  def guess_format(self):
+    return {}
+
+  def get_sample_data(self, database=None, table=None, column=None):
+    query_server = dbms.get_query_server_config(server=self.db_conf_name)
+    db = dbms.get(self.user, query_server=query_server)
+
+    assist = Assist(db)
+    response = {'status': -1}
+    sample_data = assist.get_sample_data(database, table, column)
+
+    if sample_data:
+      response['status'] = 0
+      response['headers'] = sample_data.columns
+      response['rows'] = list(sample_data.rows())
+    else:
+      response['message'] = _('Failed to get sample data.')
+
+    return response
+
+  def get_databases(self):
+    query_server = dbms.get_query_server_config(server=self.db_conf_name)
+    db = dbms.get(self.user, query_server=query_server)
+    assist = Assist(db)
+    response = {'status': -1}
+    sample_data = assist.get_databases()
+
+    if sample_data:
+      response['status'] = 0
+      response['data'] = sample_data
+    else:
+      response['message'] = _('Failed to get sample data.')
+
+    return response

+ 55 - 0
desktop/libs/indexer/src/indexer/rdbms_indexer_tests.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you 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.
+
+import json
+import logging
+
+from django.contrib.auth.models import User
+
+from nose.plugins.skip import SkipTest
+from nose.tools import assert_equal, assert_false, assert_not_equal, assert_true
+from desktop.lib.django_test_util import make_logged_in_client
+from indexer.conf import ENABLE_SQOOP
+
+from indexer.rdbms_indexer import RdbmsIndexer
+
+
+LOG = logging.getLogger(__name__)
+
+
+class TestRdbmsIndexer():
+  if not ENABLE_SQOOP.get():
+    raise SkipTest
+  def test_get_sample_data(self):
+    self.client = make_logged_in_client()
+    self.user = User.objects.get(username='test')
+
+    indexer = RdbmsIndexer(self.user, db_conf_name='mysql')
+    data = indexer.get_sample_data(database='hue', table='employee')
+
+    assert_equal(0, data['status'], data)
+    assert_true(data['rows'], data)
+
+  def test_get_databases(self):
+    self.client = make_logged_in_client()
+    self.user = User.objects.get(username='test')
+
+    indexer = RdbmsIndexer(self.user, db_conf_name='mysql')
+
+    data = indexer.get_databases()
+    assert_equal(0, data['status'], data)

+ 48 - 8
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -20,7 +20,7 @@
   from desktop import conf
   from desktop.views import commonheader, commonfooter, commonshare, commonimportexport, _ko
 
-  from indexer.conf import ENABLE_NEW_INDEXER, CONFIG_INDEXER_LIBS_PATH
+  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, CONFIG_INDEXER_LIBS_PATH
   from notebook.conf import ENABLE_SQL_INDEXER
 %>
 
@@ -228,6 +228,18 @@ ${ assist.assistPanel() }
             <!-- /ko -->
           </div>
 
+          <div class="control-group input-append" data-bind="visible: createWizard.source.inputFormat() == 'rdbms'">
+            <label for="rdbmsName" class="control-label"><div>${ _('Database') }</div>
+              <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsName" placeholder="${ _('Enter name of your database') }">
+            </label>
+          </div>
+
+          <div class="control-group input-append" data-bind="visible: createWizard.source.inputFormat() == 'rdbms'">
+            <label for="rdbmsTable" class="control-label"><div>${ _('Table') }</div>
+              <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsTable" placeholder="${ _('Enter name of your table') }">
+            </label>
+          </div>
+
           <div class="control-group" data-bind="visible: createWizard.source.inputFormat() == 'table'">
             <label for="path" class="control-label"><div>${ _('Table') }</div>
               <input type="text" class="input-xlarge" data-bind="value: createWizard.source.table, hivechooser: createWizard.source.table, skipColumns: true, apiHelperUser: '${ user }', apiHelperType: createWizard.source.apiHelperType, mainScrollable: $(MAIN_SCROLLABLE)" placeholder="${ _('Table name or <database>.<table>') }">
@@ -244,7 +256,7 @@ ${ assist.assistPanel() }
     </div>
 
     <!-- ko if: createWizard.source.show() && createWizard.source.inputFormat() != 'manual' -->
-    <div class="card step">
+    <div class="card step" data-bind="visible: createWizard.source.inputFormat() == 'file'">
       <!-- ko if: createWizard.isGuessingFormat -->
       <h4>${_('Guessing format...')} <i class="fa fa-spinner fa-spin"></i></h4>
       <!-- /ko -->
@@ -311,7 +323,7 @@ ${ assist.assistPanel() }
           <div class="control-group">
             <label for="collectionName" class="control-label "><div>${ _('Name') }</div></label>
             <!-- ko if: outputFormat() != 'table' && outputFormat() != 'database' -->
-              <input type="text" class="form-control input-xlarge" id="collectionName" data-bind="value: name, valueUpdate: 'afterkeydown'" placeholder="${ _('Name') }">
+              <input type="text" class="form-control name input-xlarge" id="collectionName" data-bind="value: name, filechooser: name, filechooserOptions: { linkMarkup: true, skipInitialPathIfEmpty: true, openOnFocus: true, selectFolder: true, uploadFile: false, uploadFolder: true}" placeholder="${ _('Name') }">
             <!-- /ko -->
 
             <!-- ko if: outputFormat() == 'table' || outputFormat() == 'database' -->
@@ -1060,11 +1072,13 @@ ${ assist.assistPanel() }
       self.inputFormatsAll = ko.observableArray([
           {'value': 'file', 'name': 'File'},
           {'value': 'manual', 'name': 'Manually'},
-          % if ENABLE_SQL_INDEXER.get():
+          % if ENABLE_SQOOP.get():
+          {'value': 'rdbms', 'name': 'External Database'},
+          % endif
+          % if ENABLE_NEW_INDEXER.get():
           {'value': 'query', 'name': 'SQL Query'},
           {'value': 'table', 'name': 'Table'},
           % endif
-          ##{'value': 'dbms', 'name': 'DBMS'},
           ##{'value': 'text', 'name': 'Paste Text'},
       ]);
       self.inputFormatsManual = ko.observableArray([
@@ -1094,6 +1108,23 @@ ${ assist.assistPanel() }
         vm.createWizard.destination.useDefaultLocation(!newVal);
       });
 
+      // Rdbms
+      self.rdbmsName = ko.observable('');
+      self.rdbmsTable = ko.observable('');
+      self.rdbmsTableName = ko.computed(function() {
+        return self.rdbmsTable().indexOf('.') > 0 ? self.rdbmsTable().split('.', 2)[1] : self.rdbmsTable();
+      });
+      self.rdbmsTableName.subscribe(function(val) {
+        if (val) {
+          vm.createWizard.guessFormat();
+        }
+        resizeElements();
+      });
+      self.rdbmsDatabaseName = ko.computed(function() {
+        return self.rdbmsTable().indexOf('.') > 0 ? self.rdbmsTable().split('.', 2)[0] : 'default';
+      });
+
+
       // Table
       self.table = ko.observable('');
       self.tableName = ko.computed(function() {
@@ -1138,6 +1169,8 @@ ${ assist.assistPanel() }
           return self.query();
         } else if (self.inputFormat() == 'manual') {
           return true;
+        }  else if (self.inputFormat() == 'rdbms') {
+          return self.rdbmsName().length > 0 && self.rdbmsTable().length > 0;
         }
       });
     };
@@ -1240,9 +1273,16 @@ ${ assist.assistPanel() }
           if (format.value == 'database' && wizard.source.inputFormat() != 'manual') {
             return false;
           }
-          else if (format.value == 'file' && wizard.source.inputFormat() != 'manual') {
+          if (format.value == 'file' && wizard.source.inputFormat() == 'manual') {
             return false;
           }
+          else if (format.value == 'file' && wizard.source.inputFormat() == 'file') {
+            return false;
+          }
+          else if (format.value == 'table' && wizard.source.inputFormat() == 'rdbms') {
+            return false;
+          }
+
           return true;
         })
       });
@@ -1460,7 +1500,7 @@ ${ assist.assistPanel() }
          );
       });
       self.readyToIndex = ko.computed(function () {
-        var validFields = self.destination.columns().length || self.destination.outputFormat() == 'database';
+        var validFields = self.destination.columns().length || self.destination.outputFormat() == 'database' || self.destination.outputFormat() == 'file';
         var validTableColumns = self.destination.outputFormat() != 'table' || ($.grep(self.destination.columns(), function(column) {
             return column.name().length == 0;
           }).length == 0
@@ -1468,7 +1508,7 @@ ${ assist.assistPanel() }
             return column.name().length == 0 || (self.source.inputFormat() != 'manual' && column.partitionValue().length == 0);
           }).length == 0
         );
-        var isTargetAlreadyExisting = !self.destination.isTargetExisting() || self.destination.outputFormat() == 'index';
+        var isTargetAlreadyExisting = ! self.destination.isTargetExisting() || self.destination.outputFormat() == 'index' || self.destination.outputFormat() == 'file';
         var isValidTable = self.destination.outputFormat() != 'table' || (
           self.destination.tableFormat() != 'kudu' || (self.destination.kuduPartitionColumns().length > 0 &&
               $.grep(self.destination.kuduPartitionColumns(), function(partition) { return partition.columns().length > 0 }).length == self.destination.kuduPartitionColumns().length && self.destination.primaryKeys().length > 0)

+ 17 - 0
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -137,6 +137,23 @@ class Notebook(object):
 
     self.data = json.dumps(_data)
 
+  def add_sqoop_snippet(self, statement, arguments, files):
+    _data = json.loads(self.data)
+
+    _data['snippets'].append(self._make_snippet({
+        u'type': u'sqoop1',
+        u'status': u'running',
+        u'properties':  {
+          u'files': files,
+          u'arguments': arguments,
+          u'archives': [],
+          u'statement': statement
+        }
+    }))
+    self._add_session(_data, 'java')
+
+    self.data = json.dumps(_data)
+
   def add_shell_snippet(self, shell_command, arguments, archives, files, env_var):
     _data = json.loads(self.data)