浏览代码

HUE-1240 [beeswax] Save a query result when using HiveServer2 interface

Save result as a table with HiveServer 2 interface
HUE-927 [beeswax] Save result from non MR query to HDFS
Backward compatible with Beeswax ctas
Will use hive INSERT OVERWRITE for saving into HDFS even when using Beeswax
Data written to the filesystem is serialized as text with columns separated
by ^A and rows separated by newlines. If any of the columns are not of
primitive type - then those columns are serialized to JSON format.
Romain Rigaux 12 年之前
父节点
当前提交
55d077a

+ 70 - 0
apps/beeswax/src/beeswax/server/dbms.py

@@ -19,6 +19,8 @@ import logging
 import thrift
 import thrift
 import time
 import time
 
 
+from django.core.urlresolvers import reverse
+from django.shortcuts import redirect
 from django.utils.encoding import force_unicode
 from django.utils.encoding import force_unicode
 from django.utils.translation import ugettext as _
 from django.utils.translation import ugettext as _
 
 
@@ -31,6 +33,7 @@ from beeswax.conf import BEESWAX_SERVER_HOST, BEESWAX_SERVER_PORT,\
   BROWSE_PARTITIONED_TABLE_LIMIT, SERVER_INTERFACE
   BROWSE_PARTITIONED_TABLE_LIMIT, SERVER_INTERFACE
 from beeswax.design import hql_query
 from beeswax.design import hql_query
 from beeswax.models import QueryHistory, HIVE_SERVER2
 from beeswax.models import QueryHistory, HIVE_SERVER2
+from desktop.lib.django_util import format_preserving_redirect
 
 
 
 
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
@@ -220,6 +223,73 @@ class Dbms:
 
 
     return self.execute_query(query, design)
     return self.execute_query(query, design)
 
 
+  def insert_query_into_directory(self, query_history, target_dir):
+    design = query_history.design.get_design()
+
+    hql = "INSERT OVERWRITE DIRECTORY '%s' %s" % (target_dir, design.query['query'])
+    return self.execute_statement(hql)
+
+
+  def create_table_as_a_select(self, request, query_history, target_table, result_meta):
+    design = query_history.design.get_design()
+    database = design.query['database']
+
+    # Case 1: Hive Server 2 backend or results straight from an existing table
+    if result_meta.in_tablename:
+      hql = 'CREATE TABLE `%s.%s` AS %s' % (database, target_table, design.query['query'])
+      #query = hql_query(hql, database=database)
+      query_history = self.execute_statement(hql)
+      url = redirect(reverse('beeswax:watch_query', args=[query_history.id]) + '?on_success_url=' + reverse('metastore:describe_table', args=[database, target_table]))
+    else:
+      # Case 2: The results are in some temporary location
+      # Beeswax backward compatibility and optimization
+      # 1. Create table
+      cols = ''
+      schema = result_meta.schema
+      for i, field in enumerate(schema.fieldSchemas):
+        if i != 0:
+          cols += ',\n'
+        cols += '`%s` %s' % (field.name, field.type)
+
+      # The representation of the delimiter is messy.
+      # It came from Java as a string, which might has been converted from an integer.
+      # So it could be "1" (^A), or "10" (\n), or "," (a comma literally).
+      delim = result_meta.delim
+      if not delim.isdigit():
+        delim = str(ord(delim))
+
+      hql = '''
+            CREATE TABLE `%s` (
+            %s
+            )
+            ROW FORMAT DELIMITED
+            FIELDS TERMINATED BY '\%s'
+            STORED AS TextFile
+            ''' % (target_table, cols, delim.zfill(3))
+
+      query = hql_query(hql)
+      self.execute_and_wait(query)
+
+      try:
+        # 2. Move the results into the table's storage
+        table_obj = self.get_table('default', target_table)
+        table_loc = request.fs.urlsplit(table_obj.path_location)[2]
+        result_dir = request.fs.urlsplit(result_meta.table_dir)[2]
+        request.fs.rename_star(result_dir, table_loc)
+        LOG.debug("Moved results from %s to %s" % (result_meta.table_dir, table_loc))
+        request.info(request, _('Saved query results as new table %(table)s') % {'table': target_table})
+        query_history.save_state(QueryHistory.STATE.expired)
+      except Exception, ex:
+        query = hql_query('DROP TABLE `%s`' % target_table)
+        try:
+          self.execute_and_wait(query)
+        except Exception, double_trouble:
+          LOG.exception('Failed to drop table "%s" as well: %s' % (target_table, double_trouble))
+        raise ex
+      url = format_preserving_redirect(request, reverse('metastore:index'))
+
+    return url
+
 
 
   def use(self, database):
   def use(self, database):
     """Beeswax interface does not support use directly."""
     """Beeswax interface does not support use directly."""

+ 11 - 0
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -522,6 +522,12 @@ class ExplainCompatible:
     self.textual = '\n'.join([line[0] for line in data_table.rows()])
     self.textual = '\n'.join([line[0] for line in data_table.rows()])
 
 
 
 
+class ResultMetaCompatible:
+
+  def __init__(self):
+    self.in_tablename = True
+
+
 class HiveServerClientCompatible:
 class HiveServerClientCompatible:
   """Same API as Beeswax"""
   """Same API as Beeswax"""
 
 
@@ -605,6 +611,11 @@ class HiveServerClientCompatible:
     return {}
     return {}
 
 
 
 
+  def get_results_metadata(self, handle):
+    # We just need to mock
+    return ResultMetaCompatible()
+
+
   def create_database(self, name, description): raise NotImplementedError()
   def create_database(self, name, description): raise NotImplementedError()
 
 
 
 

+ 8 - 9
apps/beeswax/src/beeswax/tests.py

@@ -754,7 +754,7 @@ for x in sys.stdin:
         'save': True
         'save': True
       }
       }
       resp = self.client.post('/beeswax/save_results/%s' % (qid,), save_data, follow=True)
       resp = self.client.post('/beeswax/save_results/%s' % (qid,), save_data, follow=True)
-      wait_for_query_to_finish(self.client, resp, max=60)
+      resp = wait_for_query_to_finish(self.client, resp, max=60)
 
 
       # Check that data is right
       # Check that data is right
       if verify:
       if verify:
@@ -775,24 +775,24 @@ for x in sys.stdin:
       self.cluster.fs.mkdir(TARGET_DIR_ROOT)
       self.cluster.fs.mkdir(TARGET_DIR_ROOT)
       self.cluster.fs.chown(TARGET_DIR_ROOT, user='test')
       self.cluster.fs.chown(TARGET_DIR_ROOT, user='test')
 
 
-    # Not supported. SELECT *. (Result dir is same as table dir.)
+    # SELECT *. (Result dir is same as table dir.)
     hql = "SELECT * FROM test"
     hql = "SELECT * FROM test"
     resp = _make_query(self.client, hql, wait=True, local=False, max=180.0)
     resp = _make_query(self.client, hql, wait=True, local=False, max=180.0)
     resp = save_and_verify(resp, TARGET_DIR_ROOT + '/1', verify=False)
     resp = save_and_verify(resp, TARGET_DIR_ROOT + '/1', verify=False)
-    assert_true('not supported' in resp.content)
+    # Success and went to FB
+    assert_true('File Browser' in resp.content, resp.content)
 
 
     # SELECT columns. (Result dir is in /tmp.)
     # SELECT columns. (Result dir is in /tmp.)
     hql = "SELECT foo, bar FROM test"
     hql = "SELECT foo, bar FROM test"
     resp = _make_query(self.client, hql, wait=True, local=False, max=180.0)
     resp = _make_query(self.client, hql, wait=True, local=False, max=180.0)
     resp = save_and_verify(resp, TARGET_DIR_ROOT + '/2')
     resp = save_and_verify(resp, TARGET_DIR_ROOT + '/2')
-    # Results has a link to the FB
-    assert_true('/filebrowser/view' in resp.content)
+    assert_true('File Browser' in resp.content, resp.content)
 
 
-    # Not supported. Partition tables
+    # Partition tables
     hql = "SELECT * FROM test_partitions"
     hql = "SELECT * FROM test_partitions"
     resp = _make_query(self.client, hql, wait=True, local=False, max=180.0)
     resp = _make_query(self.client, hql, wait=True, local=False, max=180.0)
     resp = save_and_verify(resp, TARGET_DIR_ROOT + '/3', verify=False)
     resp = save_and_verify(resp, TARGET_DIR_ROOT + '/3', verify=False)
-    assert_true('not supported' in resp.content)
+    assert_true('File Browser' in resp.content, resp.content)
 
 
 
 
   def test_save_results_to_tbl(self):
   def test_save_results_to_tbl(self):
@@ -810,8 +810,7 @@ for x in sys.stdin:
       wait_for_query_to_finish(self.client, resp, max=120)
       wait_for_query_to_finish(self.client, resp, max=120)
 
 
       # Check that data is right. The SELECT may not give us the whole table.
       # Check that data is right. The SELECT may not give us the whole table.
-      resp = _make_query(self.client, 'SELECT * FROM %s' % (target_tbl,), wait=True,
-                        local=False)
+      resp = _make_query(self.client, 'SELECT * FROM %s' % (target_tbl,), wait=True, local=False)
       for i in xrange(90):
       for i in xrange(90):
         assert_equal([str(i), '0x%x' % (i,)], resp.context['results'][i])
         assert_equal([str(i), '0x%x' % (i,)], resp.context['results'][i])
 
 

+ 22 - 114
apps/beeswax/src/beeswax/views.py

@@ -36,7 +36,6 @@ from desktop.lib.django_util import copy_query_dict, format_preserving_redirect,
 from desktop.lib.django_util import login_notrequired, get_desktop_uri_prefix
 from desktop.lib.django_util import login_notrequired, get_desktop_uri_prefix
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.exceptions_renderable import PopupException
 
 
-from hadoop.fs.exceptions import WebHdfsException
 from jobsub.parameterization import find_variables, substitute_variables
 from jobsub.parameterization import find_variables, substitute_variables
 
 
 import beeswax.forms
 import beeswax.forms
@@ -45,15 +44,13 @@ import beeswax.management.commands.beeswax_install_examples
 
 
 from beeswax import common, data_export, models, conf
 from beeswax import common, data_export, models, conf
 from beeswax.forms import QueryForm
 from beeswax.forms import QueryForm
-from beeswax.design import HQLdesign, hql_query
+from beeswax.design import HQLdesign
 from beeswax.models import SavedQuery, make_query_context, QueryHistory
 from beeswax.models import SavedQuery, make_query_context, QueryHistory
 from beeswax.server import dbms
 from beeswax.server import dbms
 from beeswax.server.dbms import expand_exception, get_query_server_config
 from beeswax.server.dbms import expand_exception, get_query_server_config
 
 
 
 
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
-SAVE_RESULTS_CTAS_TIMEOUT = 300         # seconds
-
 
 
 
 
 def index(request):
 def index(request):
@@ -694,147 +691,58 @@ def view_results(request, id, first_row=0):
 
 
 def save_results(request, id):
 def save_results(request, id):
   """
   """
-  Save the results of a query to an HDFS directory
+  Save the results of a query to an HDFS directory or Hive table.
   """
   """
   query_history = authorized_get_history(request, id, must_exist=True)
   query_history = authorized_get_history(request, id, must_exist=True)
 
 
+  app_name = get_app_name(request)
   server_id, state = _get_query_handle_and_state(query_history)
   server_id, state = _get_query_handle_and_state(query_history)
   query_history.save_state(state)
   query_history.save_state(state)
   error_msg, log = None, None
   error_msg, log = None, None
 
 
   if request.method == 'POST':
   if request.method == 'POST':
-    # Make sure the result is available.
-    # Note that we may still hit errors during the actual save
     if not query_history.is_success():
     if not query_history.is_success():
-      if query_history.is_failure():
-        msg = _('This query has %(state)s. Results unavailable.') % {'state': state}
-      else:
-        msg = _('The result of this query is not available yet.')
+      msg = _('This query is %(state)s. Results unavailable.') % {'state': state}
       raise PopupException(msg)
       raise PopupException(msg)
 
 
     db = dbms.get(request.user, query_history.get_query_server_config())
     db = dbms.get(request.user, query_history.get_query_server_config())
     form = beeswax.forms.SaveResultsForm(request.POST, db=db)
     form = beeswax.forms.SaveResultsForm(request.POST, db=db)
 
 
-    # Cancel goes back to results
     if request.POST.get('cancel'):
     if request.POST.get('cancel'):
-      return format_preserving_redirect(request, '/beeswax/watch/%s' % (id,))
+      return format_preserving_redirect(request, '/%s/watch/%s' % (app_name, id))
 
 
     if form.is_valid():
     if form.is_valid():
-      # Do save
-      # 1. Get the results metadata
-      assert request.POST.get('save')
       try:
       try:
         handle, state = _get_query_handle_and_state(query_history)
         handle, state = _get_query_handle_and_state(query_history)
         result_meta = db.get_results_metadata(handle)
         result_meta = db.get_results_metadata(handle)
       except Exception, ex:
       except Exception, ex:
-        LOG.exception(ex)
-        raise PopupException(_('Cannot find query.'))
-      if result_meta.table_dir:
-        result_meta.table_dir = request.fs.urlsplit(result_meta.table_dir)[2]
+        raise PopupException(_('Cannot find query: %s') % ex)
 
 
-      # 2. Check for partitioned tables
-      if result_meta.table_dir is None:
-        raise PopupException(_('Saving results from a partitioned table is not supported. You may copy from the HDFS location manually.'))
-
-      # 3. Actual saving of results
       try:
       try:
         if form.cleaned_data['save_target'] == form.SAVE_TYPE_DIR:
         if form.cleaned_data['save_target'] == form.SAVE_TYPE_DIR:
-          # To dir
-          if result_meta.in_tablename:
-            raise PopupException(_('Saving results from a query with no MapReduce jobs is not supported. '
-                                   'You may copy manually from the HDFS location %(path)s.') % {'path': result_meta.table_dir})
           target_dir = form.cleaned_data['target_dir']
           target_dir = form.cleaned_data['target_dir']
-          request.fs.rename_star(result_meta.table_dir, target_dir)
-          LOG.debug("Moved results from %s to %s" % (result_meta.table_dir, target_dir))
-          query_history.save_state(models.QueryHistory.STATE.expired)
-          return redirect(reverse('filebrowser.views.view', kwargs={'path': target_dir}))
+          query_history = db.insert_query_into_directory(query_history, target_dir)
+          redirected = redirect(reverse('beeswax:watch_query', args=[query_history.id]) \
+                                + '?on_success_url=' + reverse('filebrowser.views.view', kwargs={'path': target_dir}))
         elif form.cleaned_data['save_target'] == form.SAVE_TYPE_TBL:
         elif form.cleaned_data['save_target'] == form.SAVE_TYPE_TBL:
-          # To new table
-          try:
-            return _save_results_ctas(request, query_history, form.cleaned_data['target_table'], result_meta)
-          except Exception, bex:
-            LOG.exception(bex)
-            error_msg, log = expand_exception(bex, db)
-      except WebHdfsException, ex:
-        raise PopupException(_('The table could not be saved.'), detail=ex)
-      except IOError, ex:
-        LOG.exception(ex)
-        error_msg = str(ex)
+          redirected = db.create_table_as_a_select(request, query_history, form.cleaned_data['target_table'], result_meta)
+      except Exception, ex:
+        error_msg, log = expand_exception(ex, db)
+        raise PopupException(_('The result could not be saved: %s') % log, detail=ex)
+
+      return redirected
   else:
   else:
     form = beeswax.forms.SaveResultsForm()
     form = beeswax.forms.SaveResultsForm()
 
 
   if error_msg:
   if error_msg:
     error_msg = _('Failed to save results from query: %(error)s') % {'error': error_msg}
     error_msg = _('Failed to save results from query: %(error)s') % {'error': error_msg}
-  return render('save_results.mako', request, dict(
-    action=reverse(get_app_name(request) + ':save_results', kwargs={'id': str(id)}),
-    form=form,
-    error_msg=error_msg,
-    log=log,
-  ))
-
-
-def _save_results_ctas(request, query_history, target_table, result_meta):
-  """
-  Handle saving results as a new table. Returns HTTP response.
-  May raise Exception, IOError.
-  """
-  query_server = query_history.get_query_server_config() # Query server requires DDL support
-  db = dbms.get(request.user)
-
-  # Case 1: The results are straight from an existing table
-  if result_meta.in_tablename:
-    hql = 'CREATE TABLE `%s` AS SELECT * FROM %s' % (target_table, result_meta.in_tablename)
-    query = hql_query(hql)
-    # Display the CTAS running. Could take a long time.
-    return execute_directly(request, query, query_server, on_success_url=reverse('metastore:index'))
-
-  # Case 2: The results are in some temporary location
-  # 1. Create table
-  cols = ''
-  schema = result_meta.schema
-  for i, field in enumerate(schema.fieldSchemas):
-    if i != 0:
-      cols += ',\n'
-    cols += '`%s` %s' % (field.name, field.type)
-
-  # The representation of the delimiter is messy.
-  # It came from Java as a string, which might has been converted from an integer.
-  # So it could be "1" (^A), or "10" (\n), or "," (a comma literally).
-  delim = result_meta.delim
-  if not delim.isdigit():
-    delim = str(ord(delim))
-
-  hql = '''
-        CREATE TABLE `%s` (
-        %s
-        )
-        ROW FORMAT DELIMITED
-        FIELDS TERMINATED BY '\%s'
-        STORED AS TextFile
-        ''' % (target_table, cols, delim.zfill(3))
-
-  query = hql_query(hql)
-  db.execute_and_wait(query)
-
-  try:
-    # 2. Move the results into the table's storage
-    table_obj = db.get_table('default', target_table)
-    table_loc = request.fs.urlsplit(table_obj.path_location)[2]
-    request.fs.rename_star(result_meta.table_dir, table_loc)
-    LOG.debug("Moved results from %s to %s" % (result_meta.table_dir, table_loc))
-    messages.info(request, _('Saved query results as new table %(table)s') % {'table': target_table})
-    query_history.save_state(models.QueryHistory.STATE.expired)
-  except Exception, ex:
-    LOG.error('Error moving data into storage of table %s. Will drop table.' % (target_table,))
-    query = hql_query('DROP TABLE `%s`' % (target_table,))
-    try:
-      db.execute_directly(query)        # Don't wait for results
-    except Exception, double_trouble:
-      LOG.exception('Failed to drop table "%s" as well: %s' % (target_table, double_trouble))
-    raise ex
 
 
-  # Show tables upon success
-  return format_preserving_redirect(request, reverse('metastore:index'))
+  return render('save_results.mako', request, {
+    'action': reverse(get_app_name(request) + ':save_results', kwargs={'id': str(id)}),
+    'form': form,
+    'error_msg': error_msg,
+    'log': log,
+  })
 
 
 
 
 def confirm_query(request, query, on_success_url=None):
 def confirm_query(request, query, on_success_url=None):
@@ -1082,7 +990,7 @@ def _run_parameterized_query(request, design_id, explain):
 
 
   query_str = query_form.query.cleaned_data["query"]
   query_str = query_form.query.cleaned_data["query"]
   app_name = get_app_name(request)
   app_name = get_app_name(request)
-  query_server = get_query_server_config(app_name)  
+  query_server = get_query_server_config(app_name)
   query_type = SavedQuery.TYPES_MAPPING[app_name]
   query_type = SavedQuery.TYPES_MAPPING[app_name]
 
 
   parameterization_form_cls = make_parameterization_form(query_str)
   parameterization_form_cls = make_parameterization_form(query_str)