Эх сурвалжийг харах

HUE-2961 [editor] Backend skeleton for quick indexing of SQL data

Romain Rigaux 9 жил өмнө
parent
commit
8738a81

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

@@ -352,8 +352,6 @@ def _index(request, file_format, collection_name, query=None):
   if is_unique_generated:
     schema_fields += [{"name": unique_field, "type": "string"}]
 
-  morphline = indexer.generate_morphline_config(collection_name, file_format, unique_field)
-
   collection_manager = CollectionManagerController(request.user)
   if not collection_manager.collection_exists(collection_name):
     collection_manager.create_collection(collection_name, schema_fields, unique_key_field=unique_field)
@@ -364,7 +362,16 @@ def _index(request, file_format, collection_name, query=None):
     input_path = table_metadata.path_location
   elif file_format['inputFormat'] == 'file':
     input_path = '${nameNode}%s' % file_format["path"]
+  elif file_format['inputFormat'] == 'hs2_handle':
+    data ='aaaa'
+    searcher = CollectionManagerController(request.user)
+    columns = [field['name'] for field in collection.get('fields', [])]
+
+    searcher.update_data_from_hive(collection_name, columns, fetch_handle=file_format['fetch_handle'])
+    db.close(file_format['handle'])
   else:
     input_path = None
 
+  morphline = indexer.generate_morphline_config(collection_name, file_format, unique_field)
+
   return indexer.run_morphline(request, collection_name, morphline, input_path, query)

+ 22 - 32
desktop/libs/indexer/src/indexer/controller.py

@@ -22,6 +22,7 @@ import os
 import shutil
 
 from django.utils.translation import ugettext as _
+import tablib
 
 from desktop.lib.exceptions_renderable import PopupException
 from libsolr.api import SolrApi
@@ -266,37 +267,26 @@ class CollectionManagerController(object):
     else:
       raise PopupException(_('Could not update index. Indexing strategy %s not supported.') % indexing_strategy)
 
-  def update_data_from_hive(self, db, collection_or_core_name, database, table, columns, indexing_strategy='upload'):
-    """
-    Add hdfs path contents to index
-    """
-    # Run a custom hive query and post data to collection
-    from beeswax.server import dbms
-    import tablib
-
+  def update_data_from_hive(self, collection_or_core_name, columns, fetch_handle):
+    MAX_FETCHES = 10 # 10k rows max
+    has_more = True
     api = SolrApi(SOLR_URL.get(), self.user, SECURITY_ENABLED.get())
-    if indexing_strategy == 'upload':
-      table = db.get_table(database, table)
-      hql = "SELECT %s FROM `%s.%s` %s" % (','.join(columns), database, table.name, db._get_browse_limit_clause(table))
-      query = dbms.hql_query(hql)
-
-      try:
-        handle = db.execute_and_wait(query)
-
-        if handle:
-          result = db.fetch(handle, rows=100)
-          db.close(handle)
 
-          dataset = tablib.Dataset()
-          dataset.append(columns)
-          for row in result.rows():
-            dataset.append(row)
-
-          if not api.update(collection_or_core_name, dataset.csv, content_type='csv'):
-            raise PopupException(_('Could not update index. Check error logs for more info.'))
-        else:
-          raise PopupException(_('Could not update index. Could not fetch any data from Hive.'))
-      except Exception, e:
-        raise PopupException(_('Could not update index.'), detail=e)
-    else:
-      raise PopupException(_('Could not update index. Indexing strategy %s not supported.') % indexing_strategy)
+    try:
+      while MAX_FETCHES > 0 and has_more:
+        result = fetch_handle()
+        has_more = result['has_more']
+
+        dataset = tablib.Dataset()
+        dataset.append(columns)
+        for row in result.rows():
+          dataset.append(row)
+
+        if not api.update(collection_or_core_name, dataset.csv, content_type='csv'):
+          raise PopupException(_('Could not update index. Check error logs for more info.'))
+        
+        MAX_FETCHES -= 1
+      else:
+        raise PopupException(_('Could not update index. Could not fetch any data from Hive.'))
+    except Exception, e:
+      raise PopupException(_('Could not update index.'), detail=e)

+ 1 - 0
desktop/libs/notebook/src/notebook/views.py

@@ -202,6 +202,7 @@ def execute_and_watch(request):
     }
 
     file_format['inputFormat'] = 'hs2_handle'
+    file_format['handle'] = lambda a: get_api(request, snippet).fetch_result(notebook, snippet, 1000)
 
     job_handle = _index(request, file_format, destination, query=notebook['uuid'])
     return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_handle['handle']['id']}))