Browse Source

[importer] Create a phoenix table from a file

Missing is ready to create validation checks
Missing data loading
Romain 5 năm trước cách đây
mục cha
commit
8971c7194d

+ 22 - 0
desktop/libs/indexer/src/indexer/api3.py

@@ -54,6 +54,7 @@ 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.models import _save_pipeline
@@ -469,6 +470,27 @@ def importer_submit(request):
       destination,
       start_time
     )
+  elif destination['ouputFormat'] == 'big-table':
+    args = {
+      'source': source,
+      'destination': destination,
+      'start_time': start_time,
+      'dry_run': request.POST.get('show_command')
+    }
+    api = PhoenixIndexer(
+      request.user,
+      request.fs
+    )
+
+    job_nb = api.create_table_from_file(**args)
+
+    if request.POST.get('show_command'):
+      job_handle = {
+        'status': 0,
+        'commands': job_nb
+      }
+    else:
+      job_handle = job_nb.execute(request, batch=False)
   else:
     job_handle = _create_table(
       request,

+ 102 - 0
desktop/libs/indexer/src/indexer/indexers/phoenix_sql.py

@@ -0,0 +1,102 @@
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.import logging
+
+import logging
+import sys
+import uuid
+
+from django.urls import reverse
+from django.utils.translation import ugettext as _
+
+from notebook.models import make_notebook
+
+if sys.version_info[0] > 2:
+  from urllib.parse import urlparse, unquote as urllib_unquote
+else:
+  from urllib import unquote as urllib_unquote
+  from urlparse import urlparse
+
+
+LOG = logging.getLogger(__name__)
+
+
+class PhoenixIndexer():
+
+  def __init__(self, user, fs):
+    self.fs = fs
+    self.user = user
+
+  def create_table_from_file(self, source, destination, start_time=-1, dry_run=False):
+    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']
+    editor_type = '50'  # destination['sourceType']
+
+    columns = destination['columns']
+
+    # Until we have proper type convertion
+    for col in columns:
+      if col['type'] == 'string':
+        col['type'] = 'VARCHAR'
+
+    sql = '''CREATE TABLE IF NOT EXISTS %(table_name)s (
+%(columns)s
+CONSTRAINT my_pk PRIMARY KEY (%(primary_keys)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'))
+      }
+
+    if destination['indexerRunJob']:
+      sql += '''
+  UPSERT INTO %(table_name)s VALUES ('NY','New York',8143197);
+  UPSERT INTO %(table_name)s VALUES ('CA','Los Angeles',3844829);
+  UPSERT INTO %(table_name)s VALUES ('IL','Chicago',2842518);
+  UPSERT INTO %(table_name)s VALUES ('TX','Houston',2016582);
+  UPSERT INTO %(table_name)s VALUES ('PA','Philadelphia',1463281);
+  UPSERT INTO %(table_name)s VALUES ('AZ','Phoenix',1461575);
+  UPSERT INTO %(table_name)s VALUES ('TX','San Antonio',1256509);
+  UPSERT INTO %(table_name)s VALUES ('CA','San Diego',1255540);
+  UPSERT INTO %(table_name)s VALUES ('TX','Dallas',1213825);
+  UPSERT INTO %(table_name)s VALUES ('CA','San Jose',91233);
+  ''' % {
+          'table_name': table_name,
+        }
+
+    if dry_run:
+      return sql
+    else:
+      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
+      )

+ 25 - 8
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -531,7 +531,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
             <input type="text" class="form-control name input-xxlarge" id="collectionName" data-bind="value: name, filechooser: name, filechooserOptions: { linkMarkup: true, skipInitialPathIfEmpty: true, openOnFocus: true, selectFolder: true, displayOnlyFolders: true, uploadFile: false}" placeholder="${ _('Name') }" title="${ _('Directory must not exist in the path') }">
             <!-- /ko -->
 
-            <!-- ko if: ['index', 'stream-table'].indexOf(outputFormat()) != -1 -->
+            <!-- ko if: ['index', 'big-table', 'stream-table'].indexOf(outputFormat()) != -1 -->
             <label for="collectionName" class="control-label "><div>${ _('Name') }</div></label>
             <input type="text" class="form-control input-xlarge" id="collectionName" data-bind="value: name, valueUpdate: 'afterkeydown'" placeholder="${ _('Name') }">
             <!-- /ko -->
@@ -757,16 +757,23 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
         </div>
         <!-- /ko -->
 
-        <!-- ko if: outputFormat() == 'index' -->
+        <!-- ko if: ['index', 'big-table'].indexOf(outputFormat()) != -1 -->
         <div class="card step">
           <h4>${_('Properties')}</h4>
           <div class="card-body">
             % if ENABLE_SCALABLE_INDEXER.get():
             <div class="control-group">
               <label class="checkbox inline-block" title="${ _('Execute a cluster job to index a large dataset.') }" data-bind="visible: $root.createWizard.source.inputFormat() == 'file'">
-                <input type="checkbox" data-bind="checked: indexerRunJob"> ${_('Index with a job')}
+                <input type="checkbox" data-bind="checked: indexerRunJob">
+                  <!-- ko if: outputFormat() == 'index' -->
+                    ${_('Index with a job')}
+                  <!-- /ko -->
+                  <!-- ko if: outputFormat() == 'big-table' -->
+                    ${_('Load data')}
+                  <!-- /ko -->
               </label>
 
+              <!-- ko if: outputFormat() == 'index' -->
               <label for="path" class="control-label" data-bind="visible: indexerRunJob"><div>${ _('Libs') }</div>
                 <input type="text" class="form-control path filechooser-input input-xlarge" data-bind="value: indexerJobLibPath, filechooser: indexerJobLibPath, filechooserOptions: { linkMarkup: true, skipInitialPathIfEmpty: true }, valueUpdate: 'afterkeydown'">
               </label>
@@ -775,6 +782,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
                   <i class="fa fa-external-link-square"></i>
                 </a>
               <!-- /ko -->
+              <!-- /ko -->
             </div>
             % endif
 
@@ -784,6 +792,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
               </label>
             </div>
 
+            <!-- ko if: ['index'].indexOf(outputFormat()) != -1 -->
             <div class="control-group">
               <label for="kuduDefaultField" class="control-label"><div>${ _('Default field') }</div>
                 <select id="kuduDefaultField" data-bind="selectize: columns, selectedOptions: indexerDefaultField, selectedObjects: indexerDefaultFieldObject, optionsValue: 'name', optionsText: 'name', innerSubscriber: 'name'" size="1" multiple="false"></select>
@@ -825,6 +834,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
                 </label>
               </div>
             </span>
+            <!-- /ko -->
           </div>
         </div>
         <!-- /ko -->
@@ -935,7 +945,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
         </div>
         <!-- /ko -->
 
-        <!-- ko if: ['table', 'index', 'hbase', 'stream-table'].indexOf(outputFormat()) != -1 -->
+        <!-- ko if: ['table', 'index', 'hbase', 'big-table', 'stream-table'].indexOf(outputFormat()) != -1 -->
           <div class="card step">
             <h4>
               <!-- ko if: fieldEditorEnabled -->
@@ -1012,11 +1022,11 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
 
               <!-- ko if: $root.createWizard.source.inputFormat() !== 'manual' -->
               <form class="form-inline inline-table" data-bind="foreachVisible: { data: columns, minHeight: 54, container: MAIN_SCROLLABLE }">
-                <!-- ko if: ['table', 'stream-table'].indexOf($parent.outputFormat()) != -1 && $root.createWizard.source.inputFormat() != 'rdbms' -->
+                <!-- ko if: ['table', 'big-table', 'stream-table'].indexOf($parent.outputFormat()) != -1 && $root.createWizard.source.inputFormat() != 'rdbms' -->
                   <div data-bind="template: { name: 'table-field-template', data: $data }" class="margin-top-10 field"></div>
                 <!-- /ko -->
 
-                <!-- ko if: (['file', 'table', 'hbase'].indexOf($parent.outputFormat()) != -1 && $root.createWizard.source.inputFormat() == 'rdbms') || $parent.outputFormat() == 'index' -->
+                <!-- ko if: ['index'].indexOf($parent.outputFormat()) != -1 || (['file', 'table', 'hbase'].indexOf($parent.outputFormat()) != -1 && $root.createWizard.source.inputFormat() == 'rdbms') -->
                   <div data-bind="template: { name: 'index-field-template', data: $data }, css: { 'disabled': !keep() }" class="margin-top-10 field index-field"></div>
                 <!-- /ko -->
               </form>
@@ -2247,6 +2257,7 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
           % if ENABLE_KAFKA.get():
           {'name': '${ _("Stream Table") }', 'value': 'stream-table'},
           {'name': '${ _("Stream Topic") }', 'value': 'stream'},
+          {'name': '${ _("Phoenix Table") }', 'value': 'big-table'},
           % endif
           % if ENABLE_SQOOP.get() or ENABLE_KAFKA.get():
           {'name': '${ _("Folder") }', 'value': 'file'},
@@ -2292,6 +2303,9 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
           if (format.value === 'stream-table' && ['stream'].indexOf(wizard.source.inputFormat()) === -1) {
             return false;
           }
+          if (format.value === 'big-table' && ['file'].indexOf(wizard.source.inputFormat()) === -1) {
+            return false;
+          }
           if (format.value === 'hbase' && (wizard.source.inputFormat() !== 'rdbms' || wizard.source.rdbmsAllTablesSelected())) {
             return false;
           }
@@ -2682,11 +2696,14 @@ ${ commonheader(_("Importer"), "indexer", user, request, "60px") | n,unicode }
             || (self.source.inputFormat() === 'stream' && self.destination.outputFormat() === 'table' && self.destination.tableFormat() === 'kudu')
           ;
         }
-        var isValidTable = self.destination.outputFormat() !== 'table' || (
+        var isValidTable = ((self.destination.outputFormat() !== 'table' || (
           self.destination.tableFormat() !== 'kudu' || (
               $.grep(self.destination.kuduPartitionColumns(), function(partition) {
                 return partition.columns().length > 0
-              }).length === self.destination.kuduPartitionColumns().length && self.destination.primaryKeys().length > 0
+              }
+            ).length === self.destination.kuduPartitionColumns().length && self.destination.primaryKeys().length > 0
+          ) && (self.destination.outputFormat() !== 'big-table' || self.destination.primaryKeys().length > 0)
+        )
           )
         );
         var validIndexFields = self.destination.outputFormat() !== 'index' || ($.grep(self.destination.columns(), function(column) {

+ 1 - 1
desktop/libs/notebook/src/notebook/models.py

@@ -171,7 +171,7 @@ def make_notebook(
 
   if has_connectors():  # To improve
     data['dialect'] = interpreter['dialect']
-    data['type'] = 'flink-' + editor_connector
+    data['type'] = 'phoenix-' + editor_connector # 'flink-' + editor_connector
 
   if snippet_properties:
     data['snippets'][0]['properties'].update(snippet_properties)