Browse Source

HUE-4366 [metastore] Add an option to create a table from a file in an external location

For S3 targets, the wizard must use an s3a:// prefix as the location. As such, the fs.s3a.awsAccessKeyId and fs.s3a.awsSecretAccessKey properties must be set in core-site.xml before submitting the query.
Jenny Kim 9 years ago
parent
commit
962e026

+ 24 - 9
apps/beeswax/src/beeswax/create_table.py

@@ -23,7 +23,6 @@ import logging
 import re
 
 from django.core.urlresolvers import reverse
-from django.forms.util import ErrorDict
 from django.http import QueryDict
 from django.utils.translation import ugettext as _
 
@@ -168,10 +167,20 @@ def import_wizard(request, database='default'):
           # Go back to define columns
           do_s3_column_def, do_hive_create = True, False
 
+      load_data = s1_file_form.cleaned_data.get('load_data', 'IMPORT').upper()
+      path = s1_file_form.cleaned_data['path']
+
       #
       # Go to step 2: We've just picked the file. Preview it.
       #
       if do_s2_auto_delim:
+        if load_data == 'IMPORT':
+          if not request.fs.isfile(path):
+            raise PopupException(_('Path location must refer to a file if "Import Data" is selected.'))
+        elif load_data == 'EXTERNAL':
+          if not request.fs.isdir(path):
+            raise PopupException(_('Path location must refer to a directory if "Create External Table" is selected.'))
+
         delim_is_auto = True
         fields_list, n_cols, s2_delim_form = _delim_preview(request.fs, s1_file_form, encoding, [reader.TYPE for reader in FILE_READERS], DELIMITERS)
 
@@ -231,13 +240,16 @@ def import_wizard(request, database='default'):
       if do_hive_create:
         delim = s2_delim_form.cleaned_data['delimiter']
         table_name = s1_file_form.cleaned_data['name']
+
         proposed_query = django_mako.render_to_string("create_table_statement.mako", {
             'table': {
                 'name': table_name,
                 'comment': s1_file_form.cleaned_data['comment'],
                 'row_format': 'Delimited',
                 'field_terminator': delim,
-                'file_format': 'TextFile'
+                'file_format': 'TextFile',
+                'load_data': load_data,
+                'path': path,
              },
             'columns': [ f.cleaned_data for f in s3_col_formset.forms ],
             'partition_columns': [],
@@ -245,11 +257,8 @@ def import_wizard(request, database='default'):
             'databases': databases
           }
         )
-
-        do_load_data = s1_file_form.cleaned_data.get('do_import')
-        path = s1_file_form.cleaned_data['path']
         try:
-          return _submit_create_and_load(request, proposed_query, table_name, path, do_load_data, database=database)
+          return _submit_create_and_load(request, proposed_query, table_name, path, load_data, database=database)
         except QueryServerException, e:
           raise PopupException(_('The table could not be created.'), detail=e.message)
   else:
@@ -263,14 +272,14 @@ def import_wizard(request, database='default'):
   })
 
 
-def _submit_create_and_load(request, create_hql, table_name, path, do_load, database):
+def _submit_create_and_load(request, create_hql, table_name, path, load_data, database):
   """
-  Submit the table creation, and setup the load to happen (if ``do_load``).
+  Submit the table creation, and setup the load to happen (if ``load_data`` == IMPORT).
   """
   on_success_params = QueryDict('', mutable=True)
   app_name = get_app_name(request)
 
-  if do_load:
+  if load_data == 'IMPORT':
     on_success_params['table'] = table_name
     on_success_params['path'] = path
     on_success_params['removeHeader'] = request.POST.get('removeHeader')
@@ -296,6 +305,11 @@ def _delim_preview(fs, file_form, encoding, file_types, delimiters):
 
   path = file_form.cleaned_data['path']
   try:
+    # If path is a directory, find first file object
+    if fs.isdir(path):
+      children = fs.listdir(path)
+      if children:
+        path = '%s/%s' % (path, children[0])
     file_obj = fs.open(path)
     delim, file_type, fields_list = _parse_fields(path, file_obj, encoding, file_types, delimiters)
     file_obj.close()
@@ -477,6 +491,7 @@ def load_after_create(request, database):
 
   if is_remove_header:
     try:
+      path = path.rstrip('/')  # need to remove trailing slash before overwrite
       remove_header(request.fs, path)
     except Exception, e:
       msg = "The headers of the file could not be removed."

+ 21 - 4
apps/beeswax/src/beeswax/forms.py

@@ -20,6 +20,7 @@ from django.utils.translation import ugettext as _, ugettext_lazy as _t
 from django.core.validators import MinValueValidator, MaxValueValidator
 from django.forms import NumberInput
 
+from aws.s3 import S3_ROOT, S3A_ROOT
 from desktop.lib.django_forms import simple_formset_factory, DependencyAwareForm
 from desktop.lib.django_forms import ChoiceOrOtherField, MultiForm, SubmitButton
 from filebrowser.forms import PathField
@@ -302,15 +303,23 @@ def _clean_terminator(val):
 
 class CreateByImportFileForm(forms.Form):
   """Form for step 1 (specifying file) of the import wizard"""
+
   # Basic Data
   name = common.HiveIdentifierField(label=_t("Table Name"), required=True)
   comment = forms.CharField(label=_t("Description"), required=False)
 
   # File info
-  path = PathField(label=_t("Input File"))
-  do_import = forms.BooleanField(required=False, initial=True,
-                          label=_t("Import data from file"),
-                          help_text=_t("Automatically load this file into the table after creation."))
+  path = PathField(label=_t("Input File or Location"))
+  load_data = forms.ChoiceField(required=True,
+    choices=[
+      ("IMPORT", _("Import data")),
+      ("EXTERNAL", _("Create External Table")),
+      ("EMPTY", ("Leave Empty")),
+    ],
+    help_text=_t("Select 'import' to load data from the file into the Hive warehouse directory after creation. "
+       "Select 'external' if the table is an external table and the data files should not be moved. " +
+       "Select 'empty' if the file should only be used to define the table schema but not loaded (table will be empty).")
+  )
 
   def __init__(self, *args, **kwargs):
     self.db = kwargs.pop('db', None)
@@ -319,6 +328,14 @@ class CreateByImportFileForm(forms.Form):
   def clean_name(self):
     return _clean_tablename(self.db, self.cleaned_data['name'])
 
+  def clean_path(self):
+    path = self.cleaned_data['path'].lower()
+    if path.startswith(S3_ROOT):
+      path = path.replace(S3_ROOT, S3A_ROOT)
+    if not path.endswith('/'):
+      path = '%s/' % path
+    return path
+
 
 class CreateByImportDelimForm(forms.Form):
   """Form for step 2 (picking delimiter) of the import wizard"""

+ 3 - 3
apps/beeswax/src/beeswax/templates/create_table_statement.mako

@@ -51,7 +51,7 @@ COMMENT "${col["comment"]|n}" \
 </%def>\
 #########################
 CREATE \
-% if not table.get("use_default_location", True):
+% if table.get("load_data", "IMPORT") == 'EXTERNAL':
 EXTERNAL \
 % endif
 TABLE `${ '%s.%s' % (database, table["name"]) | n }`
@@ -90,6 +90,6 @@ ROW FORMAT \
 % if table.get("file_format") == "InputFormat":
 INPUTFORMAT ${table["input_format_class"] | n} OUTPUTFORMAT ${table["output_format_class"] | n}
 % endif
-% if not table.get("use_default_location", True):
-LOCATION "${table["external_location"] | n}"
+% if table.get("load_data", "IMPORT") == 'EXTERNAL':
+LOCATION "${table["path"] | n}"
 % endif

+ 4 - 7
apps/beeswax/src/beeswax/templates/import_wizard_choose_file.mako

@@ -150,19 +150,16 @@ ${ assist.assistPanel() }
                         )}
                         <span  class="help-inline">${unicode(file_form["path"].errors) | n}</span>
                     <span class="help-block">
-                    ${_('The HDFS path to the file on which to base this new table definition. It can be compressed (gzip) or not.')}
+                    ${_('The path to the file(s) on which to base this new table definition. It can be compressed (gzip) or not.')}
                     </span>
                     </div>
                 </div>
                 <div class="control-group">
-                  ${comps.bootstrapLabel(file_form["do_import"])}
+                  ${comps.bootstrapLabel(file_form["load_data"])}
                   <div class="controls">
-                    ${comps.field(file_form["do_import"], render_default=True)}
+                    ${comps.field(file_form["load_data"], render_default=True)}
                     <span class="help-block">
-                    ${_('Check this box to import the data in this file after creating the table definition. Leave it unchecked to define an empty table.')}
-                    <div id="fileWillBeMoved" class="alert">
-                        <strong>${_('Warning:')}</strong> ${_('The selected file is going to be moved during the import.')}
-                    </div>
+                    ${_('Select whether table data should be imported, external or empty.')}
                   </span>
                   </div>
                 </div>

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

@@ -1440,6 +1440,7 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_file': 'on',
       'path': self.cluster.fs_prefix + '/comma.dat',
+      'load_data': 'IMPORT',
       'name': 'test_create_import',
     })
     assert_equal(resp.context['fields_list'], RAW_FIELDS)
@@ -1448,6 +1449,7 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_file': 'on',
       'path': self.cluster.fs_prefix + '/comma.dat.gz',
+      'load_data': 'IMPORT',
       'name': 'test_create_import',
     })
     assert_equal(resp.context['fields_list'], RAW_FIELDS)
@@ -1456,6 +1458,7 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_preview': 'on',
       'path': self.cluster.fs_prefix + '/spacé.dat',
+      'load_data': 'IMPORT',
       'name': 'test_create_import',
       'delimiter_0': ' ',
       'delimiter_1': '',
@@ -1467,6 +1470,7 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_preview': 'on',
       'path': self.cluster.fs_prefix + '/pipes.dat',
+      'load_data': 'IMPORT',
       'name': 'test_create_import',
       'delimiter_0': '__other__',
       'delimiter_1': '|',
@@ -1478,6 +1482,7 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_preview': 'on',
       'path': self.cluster.fs_prefix + '/comma.csv',
+      'load_data': 'IMPORT',
       'name': 'test_create_import_csv',
       'delimiter_0': '__other__',
       'delimiter_1': ',',
@@ -1493,6 +1498,7 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_delim': 'on',
       'path': self.cluster.fs_prefix + '/comma.dat.gz',
+      'load_data': 'IMPORT',
       'name': 'test_create_import',
       'delimiter_0': ',',
       'delimiter_1': '',
@@ -1505,11 +1511,11 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_create': 'on',
       'path': self.cluster.fs_prefix + '/comma.dat.gz',
+      'load_data': 'IMPORT',
       'name': 'test_create_import',
       'delimiter_0': ',',
       'delimiter_1': '',
       'file_type': 'gzip',
-      'do_import': 'True',
       'cols-0-_exists': 'True',
       'cols-0-column_name': 'col_a',
       'cols-0-column_type': 'string',
@@ -1567,11 +1573,11 @@ for x in sys.stdin:
     resp = self.client.post('/beeswax/create/import_wizard/%s' % self.db_name, {
       'submit_create': 'on',
       'path': self.cluster.fs_prefix + '/comma.csv',
+      'load_data': 'IMPORT',
       'name': 'test_create_import_with_header',
       'delimiter_0': ',',
       'delimiter_1': '',
       'file_type': 'text',
-      'do_import': 'True',
       'cols-0-_exists': 'True',
       'cols-0-column_name': 'col_a',
       'cols-0-column_type': 'string',

+ 3 - 1
apps/filebrowser/src/filebrowser/api.py

@@ -46,7 +46,9 @@ def get_filesystems(request):
 
   filesystems = {}
   for k, v in FS_GETTERS.items():
-    filesystems[k] = v is not None
+    # TODO: Remove when we consolidate s3 with s3a
+    if k != 's3a':
+      filesystems[k] = v is not None
 
   response['status'] = 0
   response['filesystems'] = filesystems

+ 1 - 0
desktop/core/src/desktop/lib/fsmanager.py

@@ -33,6 +33,7 @@ DEFAULT_SCHEMA = 'hdfs'
 FS_GETTERS = {
   "hdfs": cluster.get_hdfs,
   "s3": aws.get_s3fs if is_s3_enabled() else None,
+  "s3a": aws.get_s3fs if is_s3_enabled() else None
 }
 
 

+ 2 - 1
desktop/libs/aws/src/aws/s3/__init__.py

@@ -35,8 +35,9 @@ ERRNO_MAP = {
 }
 DEFAULT_ERRNO = errno.EINVAL
 
-S3_PATH_RE = re.compile('^/*[sS]3://([^/]+)(/(.*?([^/]+)?/?))?$')
+S3_PATH_RE = re.compile('^/*[sS]3[a]?://([^/]+)(/(.*?([^/]+)?/?))?$')
 S3_ROOT = 's3://'
+S3A_ROOT = 's3a://'
 
 
 def lookup_s3error(error):