瀏覽代碼

HUE-2961 [indexer] Skeleton of quick indexing of SQL query

Romain Rigaux 9 年之前
父節點
當前提交
af1e8c1

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

@@ -363,12 +363,9 @@ def _index(request, file_format, collection_name, query=None):
   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'])
+    columns = ['_uuid'] + [field['name'] for field in file_format['columns']]
+    return searcher.update_data_from_hive(collection_name, columns, fetch_handle=file_format['fetch_handle'])
   else:
     input_path = None
 

+ 16 - 13
desktop/libs/indexer/src/indexer/controller.py

@@ -268,25 +268,28 @@ class CollectionManagerController(object):
       raise PopupException(_('Could not update index. Indexing strategy %s not supported.') % indexing_strategy)
 
   def update_data_from_hive(self, collection_or_core_name, columns, fetch_handle):
-    MAX_FETCHES = 10 # 10k rows max
+    MAX_ROWS = 10000
+    ROW_COUNT = 0
+    FETCH_BATCH = 1000
     has_more = True
     api = SolrApi(SOLR_URL.get(), self.user, SECURITY_ENABLED.get())
 
     try:
-      while MAX_FETCHES > 0 and has_more:
-        result = fetch_handle()
+      while ROW_COUNT < MAX_ROWS and has_more:
+        result = fetch_handle(FETCH_BATCH, ROW_COUNT == 0)
         has_more = result['has_more']
 
-        dataset = tablib.Dataset()
-        dataset.append(columns)
-        for row in result.rows():
-          dataset.append(row)
+        if result['data']:
+          dataset = tablib.Dataset()
+          dataset.append(columns)
+          for i, row in enumerate(result['data']):
+            dataset.append([ROW_COUNT + i] + 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.'))
+          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.'))
+
+        ROW_COUNT += len(dataset)
     except Exception, e:
       raise PopupException(_('Could not update index.'), detail=e)
+
+    return ROW_COUNT

+ 3 - 0
desktop/libs/notebook/src/notebook/connectors/hiveserver2.py

@@ -16,6 +16,7 @@
 # limitations under the License.
 
 import base64
+import binascii
 import copy
 import logging
 import re
@@ -675,6 +676,8 @@ class HS2Api(Api):
       snippet['result']['handle']['secret'], snippet['result']['handle']['guid'] = HiveServerQueryHandle.get_decoded(snippet['result']['handle']['secret'], snippet['result']['handle']['guid'])
     except KeyError:
       raise Exception('Operation has no valid handle attached')
+    except binascii.Error:
+      LOG.warn('Handle already base 64 decoded')
 
     for key in snippet['result']['handle'].keys():
       if key not in ('log_context', 'secret', 'has_result_set', 'operation_type', 'modified_row_count', 'guid'):

+ 1 - 1
desktop/libs/notebook/src/notebook/templates/notebook_ko_components.mako

@@ -192,7 +192,7 @@ except ImportError, e:
         </li>
         % if hasattr(ENABLE_NEW_INDEXER, 'get') and ENABLE_NEW_INDEXER.get():
         <li>
-          <a class="inactive-action download" href="javascript:void(0)" data-bind="click: function() { saveTarget('search-index'); savePath(''); trySaveResults(); }" title="${ _('Explore the result in an analytic dashboard') }">
+          <a class="inactive-action download" href="javascript:void(0)" data-bind="click: function() { saveTarget('search-index'); savePath('__hue__'); trySaveResults(); }" title="${ _('Explore the result in an analytic dashboard') }">
             <i class="fa fa-fw fa-area-chart"></i> ${ _('Dashboard') }
           </a>
         </li>

+ 13 - 5
desktop/libs/notebook/src/notebook/views.py

@@ -178,8 +178,11 @@ def execute_and_watch(request):
     sql, success_url = api.export_large_data_to_hdfs(notebook, snippet, destination)
     editor = make_notebook(name='Execute and watch', editor_type=editor_type, statement=sql, status='ready-execute', database=snippet['database'])
   elif action == 'index_query':
-    if not destination:
-      destination = _get_snippet_name(notebook)
+    if destination == '__hue__':
+      destination = _get_snippet_name(notebook).replace('-', '_')
+      live_indexing = True
+    else:
+      live_indexing = False
 
     sql, success_url = api.export_data_as_table(notebook, snippet, destination, is_temporary=True, location='')
     editor = make_notebook(name='Execute and watch', editor_type=editor_type, statement=sql, status='ready-execute')
@@ -201,11 +204,16 @@ def execute_and_watch(request):
         ]
     }
 
-    file_format['inputFormat'] = 'hs2_handle'
-    file_format['handle'] = lambda a: get_api(request, snippet).fetch_result(notebook, snippet, 1000)
+    if live_indexing:
+      file_format['inputFormat'] = 'hs2_handle'
+      file_format['fetch_handle'] = lambda rows, start_over: get_api(request, snippet).fetch_result(notebook, snippet, rows=rows, start_over=start_over)
 
     job_handle = _index(request, file_format, destination, query=notebook['uuid'])
-    return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_handle['handle']['id']}))
+
+    if live_indexing:
+      return redirect(reverse('search:browse', kwargs={'name': destination}))
+    else:
+      return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_handle['handle']['id']}))
   else:
     raise PopupException(_('Action %s is unknown') % action)