|
|
@@ -16,40 +16,20 @@
|
|
|
# limitations under the License.
|
|
|
|
|
|
import logging
|
|
|
-import json
|
|
|
import re
|
|
|
-import subprocess
|
|
|
-
|
|
|
-from datetime import datetime, timedelta
|
|
|
|
|
|
from django.urls import reverse
|
|
|
from django.utils.translation import ugettext as _
|
|
|
|
|
|
-from desktop.lib.exceptions_renderable import PopupException
|
|
|
from metadata.workload_analytics_client import WorkfloadAnalyticsClient
|
|
|
|
|
|
+from notebook.connectors.altus import DataEngApi as DataEngClient
|
|
|
from notebook.connectors.base import Api, QueryError
|
|
|
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
-def _exec(args):
|
|
|
- try:
|
|
|
- data = subprocess.check_output([
|
|
|
- 'altus',
|
|
|
- 'dataeng',
|
|
|
- ] +
|
|
|
- args
|
|
|
- )
|
|
|
- except Exception, e:
|
|
|
- raise PopupException(e, title=_('Error accessing'))
|
|
|
-
|
|
|
- response = json.loads(data)
|
|
|
-
|
|
|
- return response
|
|
|
-
|
|
|
-DATE_FORMAT = "%Y-%m-%d"
|
|
|
RUNNING_STATES = ('QUEUED', 'RUNNING', 'SUBMITTING')
|
|
|
|
|
|
|
|
|
@@ -63,7 +43,7 @@ class DataEngApi(Api):
|
|
|
def execute(self, notebook, snippet):
|
|
|
statement = snippet['statement']
|
|
|
|
|
|
- handle = DataEng(self.user).submit_hive_job(self.cluster_name, statement, params=None, job_xml=None)
|
|
|
+ handle = DataEngClient(self.user).submit_hive_job(self.cluster_name, statement, params=None, job_xml=None)
|
|
|
job = handle['jobs'][0]
|
|
|
|
|
|
if job['status'] not in RUNNING_STATES:
|
|
|
@@ -81,7 +61,7 @@ class DataEngApi(Api):
|
|
|
|
|
|
job_id = snippet['result']['handle']['id']
|
|
|
|
|
|
- handle = DataEng(self.user).list_jobs(job_ids=[job_id])
|
|
|
+ handle = DataEngClient(self.user).list_jobs(job_ids=[job_id])
|
|
|
job = handle['jobs'][0]
|
|
|
|
|
|
if job['status'] in RUNNING_STATES:
|
|
|
@@ -106,7 +86,7 @@ class DataEngApi(Api):
|
|
|
def cancel(self, notebook, snippet):
|
|
|
if snippet['result']['handle'].get('id'):
|
|
|
job_id = snippet['result']['handle']['id']
|
|
|
- DataEng(self.user).terminate_job(job_id=job_id)
|
|
|
+ DataEngClient(self.user).terminate_job(job_id=job_id)
|
|
|
response = {'status': 0}
|
|
|
else:
|
|
|
response = {'status': -1, 'message': _('Could not cancel because of unsuccessful submition.')}
|
|
|
@@ -145,87 +125,3 @@ class DataEngApi(Api):
|
|
|
|
|
|
def close_session(self, session):
|
|
|
pass
|
|
|
-
|
|
|
-
|
|
|
-class DataEng():
|
|
|
-
|
|
|
- def __init__(self, user): pass
|
|
|
-
|
|
|
- def list_jobs(self, submitter_crns=None, page_size=None, starting_token=None, job_statuses=None, job_ids=None, job_types=None, creation_date_before=None,
|
|
|
- creation_date_after=None, cluster_crn=None, order=None):
|
|
|
- args = ['list-jobs']
|
|
|
-
|
|
|
- if creation_date_after is None:
|
|
|
- creation_date_after = (datetime.today() - timedelta(days=7)).strftime(DATE_FORMAT)
|
|
|
-
|
|
|
- if submitter_crns:
|
|
|
- args.extend(['--submitter-crns', submitter_crns])
|
|
|
- if page_size is not None:
|
|
|
- args.extend(['--page-size', str(page_size)])
|
|
|
- if starting_token:
|
|
|
- args.extend(['--starting-token', starting_token])
|
|
|
- if job_statuses:
|
|
|
- args.extend(['--job-statuses', job_statuses])
|
|
|
- if job_ids:
|
|
|
- args.extend(['--job-ids'] + job_ids)
|
|
|
- if job_types:
|
|
|
- args.extend(['--job-types', job_types])
|
|
|
- if creation_date_before:
|
|
|
- args.extend(['--creation-date-before', creation_date_before])
|
|
|
- if creation_date_after:
|
|
|
- args.extend(['--creation-date-after', creation_date_after])
|
|
|
- if cluster_crn:
|
|
|
- args.extend(['--cluster-crn', cluster_crn])
|
|
|
- if order:
|
|
|
- args.extend(['--order', order])
|
|
|
-
|
|
|
- return _exec(args)
|
|
|
-
|
|
|
- def describe_job(self, job_id):
|
|
|
- args = ['describe-job', '--job-id', job_id]
|
|
|
-
|
|
|
- return _exec(args)
|
|
|
-
|
|
|
- def submit_hive_job(self, cluster_name, script, params=None, job_xml=None):
|
|
|
- job = {'script': script}
|
|
|
-
|
|
|
- if params:
|
|
|
- job['params'] = params
|
|
|
- if job_xml:
|
|
|
- job['jobXml'] = job_xml
|
|
|
-
|
|
|
- return self.submit_jobs(cluster_name, [{'hiveJob': job}])
|
|
|
-
|
|
|
- def submit_spark_job(self):
|
|
|
- return _exec(['submit-jobs'])
|
|
|
-
|
|
|
- def submit_yarn_job(self):
|
|
|
- return _exec(['submit-jobs'])
|
|
|
-
|
|
|
- def submit_jobs(self, cluster_name, jobs):
|
|
|
- return _exec(['submit-jobs', '--cluster-name', cluster_name, '--jobs', json.dumps(jobs)])
|
|
|
-
|
|
|
- def terminate_job(self, job_id):
|
|
|
- return _exec(['terminate-job', '--job-id', job_id])
|
|
|
-
|
|
|
-
|
|
|
- def list_clusters(self, names=None, page_size=None, starting_token=None):
|
|
|
- args = ['list-clusters']
|
|
|
-
|
|
|
- if names:
|
|
|
- args.extend(['--cluster-names', names])
|
|
|
- if page_size is not None:
|
|
|
- args.extend(['--page-size', str(page_size)])
|
|
|
- if starting_token:
|
|
|
- args.extend(['--starting-token', starting_token])
|
|
|
-
|
|
|
- return _exec(args)
|
|
|
-
|
|
|
- def create_cluster(self):
|
|
|
- return _exec(['create-cluster'])
|
|
|
-
|
|
|
- def delete_cluster(self):
|
|
|
- return _exec(['delete-cluster'])
|
|
|
-
|
|
|
- def describe_clusters(self):
|
|
|
- return _exec(['describe-cluster'])
|