Forráskód Böngészése

[Importer] adding direct upload to importer

ayush.goyal 4 éve
szülő
commit
e411d174c0

+ 3 - 0
desktop/conf.dist/hue.ini

@@ -1647,6 +1647,9 @@
   # Flag to turn on Kafka topic ingest.
   ## enable_kafka=false
 
+  # Flag to turn on the direct upload of a small file.
+  ## enable_direct_upload=false
+
 
 ###########################################################################
 # Settings to configure Job Designer

+ 3 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -1633,6 +1633,9 @@
   # Flag to turn on Kafka topic ingest.
   ## enable_kafka=false
 
+  # Flag to turn on the direct upload of a small file.
+  ## enable_direct_upload=false
+
 
 ###########################################################################
 # Settings to configure Job Designer

+ 2 - 1
desktop/core/src/desktop/settings.py

@@ -117,7 +117,8 @@ USE_TZ = False
 # URL that handles the media served from MEDIA_ROOT. Make sure to use a
 # trailing slash.
 # Examples: "http://media.lawrence.com/media/", "http://example.com/media/"
-MEDIA_URL = ''
+MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
+MEDIA_URL = '/media/'
 
 
 ############################################################

+ 70 - 10
desktop/libs/indexer/src/indexer/api3.py

@@ -20,6 +20,7 @@ standard_library.install_aliases()
 
 from builtins import zip
 from past.builtins import basestring
+import csv
 import json
 import logging
 import urllib.error
@@ -27,6 +28,7 @@ import sys
 
 from django.urls import reverse
 from django.views.decorators.http import require_POST
+from django.core.files.storage import FileSystemStorage
 
 LOG = logging.getLogger(__name__)
 
@@ -48,14 +50,14 @@ from notebook.models import MockedDjangoRequest, escape_rows
 
 from indexer.controller import CollectionManagerController
 from indexer.file_format import HiveFormat
-from indexer.fields import Field
+from indexer.fields import Field, guess_field_type_from_samples
 from indexer.indexers.envelope import _envelope_job
 from indexer.indexers.base import get_api
 from indexer.indexers.flink_sql import FlinkIndexer
 from indexer.indexers.morphline import MorphlineIndexer, _create_solr_collection
 from indexer.indexers.phoenix_sql import PhoenixIndexer
 from indexer.indexers.rdbms import run_sqoop, _get_api
-from indexer.indexers.sql import _create_database, _create_table
+from indexer.indexers.sql import _create_database, _create_table, _create_table_from_local
 from indexer.models import _save_pipeline
 from indexer.solr_client import SolrClient, MAX_UPLOAD_SIZE
 from indexer.indexers.flume import FlumeIndexer
@@ -86,6 +88,7 @@ try:
 except ImportError as e:
   LOG.warning('Solr Search interface is not enabled')
 
+csv_data = []
 
 def _escape_white_space_characters(s, inverse=False):
   MAPPINGS = {
@@ -117,7 +120,16 @@ def _convert_format(format_dict, inverse=False):
 def guess_format(request):
   file_format = json.loads(request.POST.get('fileFormat', '{}'))
 
-  if file_format['inputFormat'] == 'file':
+  if file_format['inputFormat'] == 'localfile':
+    format_ = {
+      "quoteChar": "\"",
+      "recordSeparator": '\\n',
+      "type": "csv",
+      "hasHeader": True,
+      "fieldSeparator": ","
+    }
+
+  elif file_format['inputFormat'] == 'file':
     path = urllib_unquote(file_format["path"])
     indexer = MorphlineIndexer(request.user, request.fs)
     if not request.fs.isfile(path):
@@ -208,11 +220,50 @@ def guess_format(request):
   format_['status'] = 0
   return JsonResponse(format_)
 
+def decode_utf8(input_iterator):
+  for l in input_iterator:
+    yield l.decode('utf-8')
 
 def guess_field_types(request):
   file_format = json.loads(request.POST.get('fileFormat', '{}'))
 
-  if file_format['inputFormat'] == 'file':
+  if file_format['inputFormat'] == 'localfile':
+    upload_file = request.FILES['inputfile']
+    fs = FileSystemStorage()
+    name = fs.save(upload_file.name, upload_file)
+    reader = csv.reader(decode_utf8(upload_file))
+
+    sample = []
+    column_row = []
+
+    for count, row in enumerate(reader):
+      if count == 0:
+        column_row = row
+      elif count <= 5:
+        sample.append(row)
+        csv_data.append(row)
+      else:
+        csv_data.append(row)
+
+    field_type_guesses = []
+    for col in range(len(column_row)):
+      column_samples = [sample_row[col] for sample_row in sample if len(sample_row) > col]
+
+      field_type_guess = guess_field_type_from_samples(column_samples)
+      field_type_guesses.append(field_type_guess)
+
+    columns = [
+      Field(column_row[i], field_type_guesses[i]).to_dict()
+      for i in range(len(column_row))
+    ]
+
+    format_ = {
+      'file_url': fs.url(name),
+      'columns': columns,
+      'sample': sample
+    }
+
+  elif file_format['inputFormat'] == 'file':
     indexer = MorphlineIndexer(request.user, request.fs)
     path = urllib_unquote(file_format["path"])
     stream = request.fs.open(path)
@@ -493,12 +544,21 @@ def importer_submit(request):
     else:
       job_handle = job_nb.execute(request, batch=False)
   else:
-    job_handle = _create_table(
-      request,
-      source,
-      destination,
-      start_time
-    )
+    if source['inputFormat'] == 'localfile':
+      job_handle = _create_table_from_local(
+        request,
+        source,
+        destination,
+        csv_data,
+        start_time
+      )
+    else:
+      job_handle = _create_table(
+        request,
+        source,
+        destination,
+        start_time
+      )
 
   request.audit = {
     'operation': 'EXPORT',

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

@@ -144,6 +144,13 @@ ENABLE_ALTUS = Config(
   default=False
 )
 
+ENABLE_DIRECT_UPLOAD = Config(
+  key="enable_direct_upload",
+  help=_t("Flag to turn on the direct upload of a small file."),
+  type=bool,
+  default=False
+)
+
 # Unused
 BATCH_INDEXER_PATH = Config(
   key="batch_indexer_path",

+ 54 - 0
desktop/libs/indexer/src/indexer/indexers/sql.py

@@ -266,7 +266,53 @@ class SQLIndexer(object):
         is_task=True
     )
 
+  def create_table_from_local_file(self, source, destination, csv_data, start_time=-1):
+    if '.' in destination['name']:
+      database, table_name = destination['name'].split('.', 1)
+    else:
+      database = 'default'
+      table_name = destination['name']
+    final_table_name = table_name
+
+    # source_type = source['sourceType']
+    source_type = 'hive'
+    # editor_type = '50'  # destination['sourceType']
+    # editor_type = destination['sourceType']
+    editor_type = 'hive'
+
+    columns = destination['columns']
+
+    sql = '''CREATE TABLE IF NOT EXISTS %(table_name)s (
+%(columns)s
+);
+''' % {
+          'database': database,
+          'table_name': table_name,
+          'columns': ',\n'.join(['  `%(name)s` %(type)s' % col for col in columns]),
+          'primary_keys': ', '.join(destination.get('indexerPrimaryKey'))
+      }
+
+    for csv_row in csv_data:
+      sql += '''
+          INSERT INTO %(table_name)s VALUES %(csv_row)s;
+          ''' % {
+                  'table_name': table_name,
+                  'csv_row': tuple(csv_row)
+                }
+
+    on_success_url = reverse('metastore:describe_table', kwargs={'database': database, 'table': final_table_name}) + \
+        '?source_type=' + source_type
 
+    return make_notebook(
+        name=_('Creating table %(database)s.%(table)s') % {'database': database, 'table': final_table_name},
+        editor_type=editor_type,
+        statement=sql.strip(),
+        status='ready',
+        database=database,
+        on_success_url=on_success_url,
+        last_executed=start_time,
+        is_task=True
+    )
 
 def _create_database(request, source, destination, start_time):
   database = destination['name']
@@ -311,3 +357,11 @@ def _create_table(request, source, destination, start_time=-1):
     return {'status': 0, 'commands': notebook.get_str()}
   else:
     return notebook.execute(request, batch=False)
+
+def _create_table_from_local(request, source, destination, csv_data, start_time=-1):
+  notebook = SQLIndexer(user=request.user, fs=request.fs).create_table_from_local_file(source, destination, csv_data, start_time)
+
+  if request.POST.get('show_command'):
+    return {'status': 0, 'commands': notebook.get_str()}
+  else:
+    return notebook.execute(request, batch=False)

+ 70 - 5
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -24,7 +24,7 @@
   from impala import impala_flags
   from notebook.conf import ENABLE_SQL_INDEXER
 
-  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, ENABLE_KAFKA, CONFIG_INDEXER_LIBS_PATH, ENABLE_SCALABLE_INDEXER, ENABLE_ALTUS, ENABLE_ENVELOPE, ENABLE_FIELD_EDITOR
+  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, ENABLE_KAFKA, CONFIG_INDEXER_LIBS_PATH, ENABLE_SCALABLE_INDEXER, ENABLE_ALTUS, ENABLE_ENVELOPE, ENABLE_FIELD_EDITOR, ENABLE_DIRECT_UPLOAD
 
   if sys.version_info[0] > 2:
     from django.utils.translation import gettext as _
@@ -234,6 +234,17 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
             <!-- /ko -->
           </div>
 
+          <div data-bind="visible: createWizard.source.inputFormat() == 'localfile'">
+              <form method="post" action="" enctype="multipart/form-data" id="uploadform">
+                <div >
+                    <input type="file" id="inputfile" name="inputfile" accept=".csv">
+                    <input type="button" class="button" value="Upload" id="but_upload">
+                </div>
+                <label for="path" class="control-label"><div>${ _('Path') }</div>
+                  <input type="text" id="file_path" data-bind="value: createWizard.source.path">
+            </form>
+          </div>
+
           <!-- ko if: createWizard.source.inputFormat() == 'rdbms' -->
 
             <!-- ko if: createWizard.source.rdbmsMode() -->
@@ -441,7 +452,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
     </div>
 
     <!-- ko if: createWizard.source.show() && createWizard.source.inputFormat() != 'manual' -->
-    <div class="card step" data-bind="visible: createWizard.source.inputFormat() == 'file' || createWizard.source.inputFormat() == 'stream'">
+    <div class="card step" data-bind="visible: createWizard.source.inputFormat() == 'file' || createWizard.source.inputFormat() == 'localfile' || createWizard.source.inputFormat() == 'stream'">
       <!-- ko if: createWizard.isGuessingFormat -->
       <h4>${_('Guessing format...')} <i class="fa fa-spinner fa-spin"></i></h4>
       <!-- /ko -->
@@ -449,7 +460,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
       <h4>${_('Format')}</h4>
       <div class="card-body">
         <label data-bind="visible: (createWizard.prefill.source_type().length == 0 || createWizard.prefill.target_type() == 'index') &&
-            (createWizard.source.inputFormat() == 'file' || createWizard.source.inputFormat() == 'stream')">
+            (createWizard.source.inputFormat() == 'file' || createWizard.source.inputFormat() == 'localfile' || createWizard.source.inputFormat() == 'stream')">
           <div>${_('File Type')}</div>
           <select data-bind="selectize: $root.createWizard.fileTypes, value: $root.createWizard.fileTypeName,
               optionsText: 'description', optionsValue: 'name'"></select>
@@ -1760,7 +1771,10 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
           % if ENABLE_ENVELOPE.get():
           {'value': 'connector', 'name': 'Connectors'},
           % endif
-          {'value': 'manual', 'name': 'Manually'}
+          {'value': 'manual', 'name': 'Manually'},
+          % if ENABLE_DIRECT_UPLOAD.get():
+          {'value': 'localfile', 'name': 'LocalFile'}
+          % endif
           ##{'value': 'text', 'name': 'Paste Text'},
       ]);
       self.inputFormatsManual = ko.observableArray([
@@ -1776,7 +1790,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
       // File
       self.path = ko.observable('');
       self.path.subscribe(function(val) {
-        if (val) {
+        if (val && self.inputFormat() != 'localfile') {
           wizard.guessFormat();
           wizard.destination.nonDefaultLocation(val);
         }
@@ -2136,6 +2150,9 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
       });
 
       self.show = ko.pureComputed(function() {
+        if (self.inputFormat() === 'localfile') {
+          return self.path().length > 0;
+        }
         if (self.inputFormat() === 'file') {
           return self.path().length > 0;
         }
@@ -3139,6 +3156,54 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
 
       $(window).on('resize', resizeElements);
 
+      document.getElementById('inputfile').onchange = function () {
+        upload();
+        };
+        function upload() {
+          var fd = new FormData();
+          fd.append('fileFormat', ko.mapping.toJSON(viewModel.createWizard.source));
+            $.ajax({
+              url:"/indexer/api/indexer/guess_format",
+              type: 'post',
+              data: fd,
+              contentType:false,
+              cache: false,
+              processData:false,
+              beforeSend: function (){
+                viewModel.createWizard.isGuessingFormat(true);
+              },
+              success:function (resp) {
+                var newFormat = ko.mapping.fromJS(new FileType(resp['type'], resp));
+                viewModel.createWizard.source.format(newFormat);
+                viewModel.createWizard.isGuessingFormat(false);
+              }
+            });
+        };
+
+      $("#but_upload").click(function() {
+          var fd = new FormData();
+          var files = $('#inputfile')[0].files[0];
+          fd.append('inputfile', files);
+          fd.append('fileFormat', ko.mapping.toJSON(viewModel.createWizard.source));
+  
+          $.ajax({
+              url: '/indexer/api/indexer/guess_field_types',
+              type: 'post',
+              data: fd,
+              contentType: false,
+              processData: false,
+              beforeSend: function(){
+                viewModel.createWizard.isGuessingFieldTypes(true);
+              },
+              success: function(response){
+                viewModel.createWizard.loadSampleData(response);
+                viewModel.createWizard.isGuessingFieldTypes(false);
+                viewModel.createWizard.source.path(response['file_url']);
+                document.getElementById("file_path").value = response['file_url'];
+              },
+          });
+      });
+
       $('.importer-droppable').droppable({
         accept: ".draggableText",
         drop: function (e, ui) {