Browse Source

HUE-8473 [cluster] Support submitting Spark jobs to remote cluster

Romain Rigaux 7 years ago
parent
commit
e7be4fbc

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

@@ -152,7 +152,7 @@ def get_context_computes(request, interface):
         } for cluster in AnalyticDbApi(request.user).list_clusters()['clusters'] if cluster.get('status') == 'CREATED']
         } for cluster in AnalyticDbApi(request.user).list_clusters()['clusters'] if cluster.get('status') == 'CREATED']
       )
       )
 
 
-  if interface == 'oozie' or interface == 'jobs':
+  if interface == 'oozie' or interface == 'jobs' or interface == 'spark2':
     if [cluster for cluster in clusters if cluster['type'] == 'altus']:
     if [cluster for cluster in clusters if cluster['type'] == 'altus']:
       computes.extend([{
       computes.extend([{
           'id': cluster.get('crn'),
           'id': cluster.get('crn'),
@@ -164,6 +164,7 @@ def get_context_computes(request, interface):
           'type': 'altus-de'
           'type': 'altus-de'
         } for cluster in DataEngApi(request.user).list_clusters()['clusters']]
         } for cluster in DataEngApi(request.user).list_clusters()['clusters']]
       )
       )
+      # TODO if interface == 'spark2' keep only SPARK type
 
 
   response[interface] = computes
   response[interface] = computes
   response['status'] = 0
   response['status'] = 0

+ 12 - 2
desktop/libs/notebook/src/notebook/connectors/altus.py

@@ -22,6 +22,7 @@ from datetime import datetime,  timedelta
 
 
 from django.urls import reverse
 from django.urls import reverse
 from django.utils.translation import ugettext as _
 from django.utils.translation import ugettext as _
+
 from metadata.conf import ALTUS
 from metadata.conf import ALTUS
 from navoptapi.api_lib import ApiLib
 from navoptapi.api_lib import ApiLib
 
 
@@ -125,8 +126,17 @@ class DataEngApi():
 
 
     return self.submit_jobs(cluster_name, [{'hiveJob': job}])
     return self.submit_jobs(cluster_name, [{'hiveJob': job}])
 
 
-  def submit_spark_job(self):
-    return _exec('dataeng', 'submitJobs')
+  def submit_spark_job(self, cluster_name, jars=None, main_class=None, arguments=None, spark_arguments=None, properties_file=None):
+    job = {
+      "jars": jars if jars else [],
+      "applicationArguments": arguments if arguments else [],
+      #"sparkArguments": "string",
+      #"propertiesFile": "string"
+    }
+    if main_class:
+      job["mainClass"] = main_class
+
+    return self.submit_jobs(cluster_name, [{'sparkJob': job}])
 
 
   def submit_yarn_job(self):
   def submit_yarn_job(self):
     return _exec('dataeng', 'submitJobs')
     return _exec('dataeng', 'submitJobs')

+ 3 - 1
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -311,6 +311,8 @@ def get_api(request, snippet):
     cluster = snippet.get('compute').get('id')
     cluster = snippet.get('compute').get('id')
   if cluster and 'crn:altus:dataware:' in cluster:
   if cluster and 'crn:altus:dataware:' in cluster:
     interface = 'altus-adb'
     interface = 'altus-adb'
+  elif cluster and 'crn:altus:dataeng:' in cluster:
+    interface = 'dataeng'
   if cluster:
   if cluster:
     LOG.info('Selected cluster %s' % cluster)
     LOG.info('Selected cluster %s' % cluster)
 
 
@@ -336,7 +338,7 @@ def get_api(request, snippet):
     return AltusAdbApi(user=request.user, cluster_name=cluster, request=request)
     return AltusAdbApi(user=request.user, cluster_name=cluster, request=request)
   elif interface == 'dataeng':
   elif interface == 'dataeng':
     from notebook.connectors.dataeng import DataEngApi
     from notebook.connectors.dataeng import DataEngApi
-    return DataEngApi(user=request.user, request=request, cluster_name=cluster.get('name'))
+    return DataEngApi(user=request.user, request=request, cluster_name=cluster)
   elif interface == 'jdbc' or interface == 'teradata':
   elif interface == 'jdbc' or interface == 'teradata':
     from notebook.connectors.jdbc import JdbcApi
     from notebook.connectors.jdbc import JdbcApi
     return JdbcApi(request.user, interpreter=interpreter)
     return JdbcApi(request.user, interpreter=interpreter)

+ 15 - 4
desktop/libs/notebook/src/notebook/connectors/dataeng.py

@@ -23,7 +23,7 @@ from django.utils.translation import ugettext as _
 
 
 from metadata.workload_analytics_client import WorkfloadAnalyticsClient
 from metadata.workload_analytics_client import WorkfloadAnalyticsClient
 
 
-from notebook.connectors.altus import DataEngApi
+from notebook.connectors.altus import DataEngApi as AltusDataEngApi
 from notebook.connectors.base import Api, QueryError
 from notebook.connectors.base import Api, QueryError
 
 
 
 
@@ -43,7 +43,18 @@ class DataEngApi(Api):
   def execute(self, notebook, snippet):
   def execute(self, notebook, snippet):
     statement = snippet['statement']
     statement = snippet['statement']
 
 
-    handle = DataEngApi(self.user).submit_hive_job(self.cluster_name, statement, params=None, job_xml=None)
+    if snippet['type'] == 'spark2':
+      handle = AltusDataEngApi(self.user).submit_spark_job(
+          cluster_name=self.cluster_name,
+          jars=snippet['properties']['jars'],
+          main_class=snippet['properties']['class'],
+          arguments=snippet['properties']['spark_arguments'],
+#           spark_arguments,
+#           properties_file
+      )
+    else:
+      handle = AltusDataEngApi(self.user).submit_hive_job(self.cluster_name, statement, params=None, job_xml=None)
+
     job = handle['jobs'][0]
     job = handle['jobs'][0]
 
 
     if job['status'] not in RUNNING_STATES:
     if job['status'] not in RUNNING_STATES:
@@ -61,7 +72,7 @@ class DataEngApi(Api):
 
 
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
 
 
-    handle = DataEngApi(self.user).list_jobs(job_ids=[job_id])
+    handle = AltusDataEngApi(self.user).list_jobs(job_ids=[job_id])
     job = handle['jobs'][0]
     job = handle['jobs'][0]
 
 
     if job['status'] in RUNNING_STATES:
     if job['status'] in RUNNING_STATES:
@@ -86,7 +97,7 @@ class DataEngApi(Api):
   def cancel(self, notebook, snippet):
   def cancel(self, notebook, snippet):
     if snippet['result']['handle'].get('id'):
     if snippet['result']['handle'].get('id'):
       job_id = snippet['result']['handle']['id']
       job_id = snippet['result']['handle']['id']
-      DataEngApi(self.user).terminate_job(job_id=job_id)
+      AltusDataEngApi(self.user).terminate_job(job_id=job_id)
       response = {'status': 0}
       response = {'status': 0}
     else:
     else:
       response = {'status': -1, 'message': _('Could not cancel because of unsuccessful submition.')}
       response = {'status': -1, 'message': _('Could not cancel because of unsuccessful submition.')}

+ 14 - 11
desktop/libs/notebook/src/notebook/templates/editor_components.mako

@@ -978,16 +978,17 @@ ${ sqlSyntaxDropdown.sqlSyntaxDropdown() }
 </script>
 </script>
 
 
 <script type="text/html" id="snippet-header-database-selection">
 <script type="text/html" id="snippet-header-database-selection">
-  <!-- ko if: loadingContext() && isSqlDialect()  -->
+  <!-- ko if: loadingContext() && (isSqlDialect() || type() == 'spark2') -->
   <i class="fa fa-spinner fa-spin muted"></i>
   <i class="fa fa-spinner fa-spin muted"></i>
   <!-- /ko -->
   <!-- /ko -->
-  <!-- ko if: !loadingContext() && isSqlDialect() -->
+
+  <!-- ko if: !loadingContext() && (isSqlDialect() || type() == 'spark2') -->
   <!-- ko if: availableComputes().length > 1 || namespaceRefreshEnabled() -->
   <!-- ko if: availableComputes().length > 1 || namespaceRefreshEnabled() -->
-  <!-- ko if: availableComputes().length > 0-->
+  <!-- ko if: availableComputes().length > 0 -->
   <span class="editor-header-title">${ _('Compute') }</span>
   <span class="editor-header-title">${ _('Compute') }</span>
   <div data-bind="component: { name: 'hue-drop-down', params: { value: compute, entries: availableComputes, labelAttribute: 'name', searchable: true, linkTitle: '${ _ko('Active compute') }' } }" style="display: inline-block"></div>
   <div data-bind="component: { name: 'hue-drop-down', params: { value: compute, entries: availableComputes, labelAttribute: 'name', searchable: true, linkTitle: '${ _ko('Active compute') }' } }" style="display: inline-block"></div>
   <!-- /ko -->
   <!-- /ko -->
-  <!-- ko if: availableComputes().length === 0-->
+  <!-- ko if: availableComputes().length === 0 -->
   <span class="editor-header-title"><i class="fa fa-warning"></i> ${ _('No computes found') }</span>
   <span class="editor-header-title"><i class="fa fa-warning"></i> ${ _('No computes found') }</span>
   <!-- /ko -->
   <!-- /ko -->
   <!-- /ko -->
   <!-- /ko -->
@@ -997,13 +998,15 @@ ${ sqlSyntaxDropdown.sqlSyntaxDropdown() }
   <div data-bind="component: { name: 'hue-drop-down', params: { value: namespace, entries: availableNamespaces, labelAttribute: 'name', searchable: true, linkTitle: '${ _ko('Active namespace') }' } }" style="display: inline-block"></div>
   <div data-bind="component: { name: 'hue-drop-down', params: { value: namespace, entries: availableNamespaces, labelAttribute: 'name', searchable: true, linkTitle: '${ _ko('Active namespace') }' } }" style="display: inline-block"></div>
   <!-- /ko -->
   <!-- /ko -->
 
 
-  <!-- ko if: availableDatabases().length > 0 -->
-  <span class="editor-header-title">${ _('Database') }</span>
-  <div data-bind="component: { name: 'hue-drop-down', params: { value: database, entries: availableDatabases, foreachVisible: true, searchable: true, linkTitle: '${ _ko('Active database') }' } }" style="display: inline-block"></div>
-  <!-- /ko -->
-  <!-- ko if: availableDatabases().length === 0-->
-  <span class="editor-header-title"><i class="fa fa-warning"></i> ${ _('No databases found') }</span>
-  <!-- /ko -->
+  <!-- ko if: isSqlDialect() -->
+    <!-- ko if: availableDatabases().length > 0 -->
+    <span class="editor-header-title">${ _('Database') }</span>
+    <div data-bind="component: { name: 'hue-drop-down', params: { value: database, entries: availableDatabases, foreachVisible: true, searchable: true, linkTitle: '${ _ko('Active database') }' } }" style="display: inline-block"></div>
+    <!-- /ko -->
+    <!-- ko if: availableDatabases().length === 0 -->
+    <span class="editor-header-title"><i class="fa fa-warning"></i> ${ _('No databases found') }</span>
+    <!-- /ko -->
+    <!-- /ko -->
   <!-- /ko -->
   <!-- /ko -->
 </script>
 </script>