| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241 |
- #!/usr/bin/env python
- # 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 json
- import logging
- import time
- from django.core.urlresolvers import reverse
- from django.utils.translation import ugettext as _
- from desktop.lib.i18n import smart_str
- from desktop.lib.view_util import format_duration_in_millis
- from liboozie.oozie_api import get_oozie
- from oozie.models import Workflow, Pig
- from oozie.views.api import get_log as get_workflow_logs
- from oozie.views.editor import _submit_workflow
- LOG = logging.getLogger(__name__)
- def get(fs, jt, user):
- return OozieApi(fs, jt, user)
- class OozieApi(object):
- """
- Oozie submission.
- """
- WORKFLOW_NAME = 'pig-app-hue-script'
- LOG_START_PATTERN = '(Pig script \[(?:[\w.-]+)\] content:.+)'
- LOG_END_PATTERN = '(<<< Invocation of Pig command completed <<<|<<< Invocation of Main class completed <<<)'
- MAX_DASHBOARD_JOBS = 100
- def __init__(self, fs, jt, user):
- self.oozie_api = get_oozie(user)
- self.fs = fs
- self.jt = jt
- self.user = user
- def submit(self, pig_script, params):
- workflow = None
- try:
- workflow = self._create_workflow(pig_script, params)
- mapping = dict([(param['name'], param['value']) for param in workflow.get_parameters()])
- oozie_wf = _submit_workflow(self.user, self.fs, self.jt, workflow, mapping)
- finally:
- if workflow:
- workflow.delete(skip_trash=True)
- return oozie_wf
- def _create_workflow(self, pig_script, params):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.schema_version = 'uri:oozie:workflow:0.5'
- workflow.name = OozieApi.WORKFLOW_NAME
- workflow.is_history = True
- if pig_script.use_hcatalog:
- workflow.add_parameter("oozie.action.sharelib.for.pig", "pig,hcatalog")
- workflow.save()
- Workflow.objects.initialize(workflow, self.fs)
- script_path = workflow.deployment_dir + '/script.pig'
- if self.fs: # For testing, difficult to mock
- self.fs.do_as_user(self.user.username, self.fs.create, script_path, data=smart_str(pig_script.dict['script']))
- files = []
- archives = []
- popup_params = json.loads(params)
- popup_params_names = [param['name'] for param in popup_params]
- pig_params = self._build_parameters(popup_params)
- if pig_script.isV2:
- pig_params += [{"type": "argument", "value": param} for param in pig_script.dict['parameters']]
- job_properties = [{"name": prop.split('=', 1)[0], "value": prop.split('=', 1)[1]} for prop in pig_script.dict['hadoopProperties']]
- for resource in pig_script.dict['resources']:
- if resource.endswith('.zip') or resource.endswith('.tgz') or resource.endswith('.tar') or resource.endswith('.gz'):
- archives.append({"dummy": "", "name": resource})
- else:
- files.append(resource)
- else:
- script_params = [param for param in pig_script.dict['parameters'] if param['name'] not in popup_params_names]
- pig_params += self._build_parameters(script_params)
- job_properties = [{"name": prop['name'], "value": prop['value']} for prop in pig_script.dict['hadoopProperties']]
- for resource in pig_script.dict['resources']:
- if resource['type'] == 'file':
- files.append(resource['value'])
- if resource['type'] == 'archive':
- archives.append({"dummy": "", "name": resource['value']})
- action = Pig.objects.create(
- name='pig-5760',
- script_path=script_path,
- workflow=workflow,
- node_type='pig',
- params=json.dumps(pig_params),
- files=json.dumps(files),
- archives=json.dumps(archives),
- job_properties=json.dumps(job_properties)
- )
- credentials = []
- if pig_script.use_hcatalog and self.oozie_api.security_enabled:
- credentials.append({'name': 'hcat', 'value': True})
- if pig_script.use_hbase and self.oozie_api.security_enabled:
- credentials.append({'name': 'hbase', 'value': True})
- if credentials:
- action.credentials = credentials # Note, action.credentials is a @setter here
- action.save()
- action.add_node(workflow.end)
- start_link = workflow.start.get_link()
- start_link.child = action
- start_link.save()
- return workflow
- def _build_parameters(self, params):
- pig_params = []
- for param in params:
- if param['name'].startswith('-'):
- pig_params.append({"type": "argument", "value": "%(name)s" % param})
- if param['value']:
- pig_params.append({"type": "argument", "value": "%(value)s" % param})
- else:
- # Simpler way and backward compatibility for parameters
- pig_params.append({"type": "argument", "value": "-param"})
- pig_params.append({"type": "argument", "value": "%(name)s=%(value)s" % param})
- return pig_params
- def stop(self, job_id):
- return self.oozie_api.job_control(job_id, 'kill')
- def get_jobs(self):
- kwargs = {'cnt': OozieApi.MAX_DASHBOARD_JOBS,}
- kwargs['filters'] = [
- ('user', self.user.username),
- ('name', OozieApi.WORKFLOW_NAME)
- ]
- return self.oozie_api.get_workflows(**kwargs).jobs
- def get_log(self, request, oozie_workflow, make_links=True):
- return get_workflow_logs(request, oozie_workflow, make_links=make_links, log_start_pattern=self.LOG_START_PATTERN,
- log_end_pattern=self.LOG_END_PATTERN)
- def massaged_jobs_for_json(self, request, oozie_jobs, hue_jobs):
- jobs = []
- hue_jobs = dict([(script.dict.get('job_id'), script) for script in hue_jobs if script.dict.get('job_id')])
- for job in oozie_jobs:
- if job.is_running():
- job = self.oozie_api.get_job(job.id)
- get_copy = request.GET.copy() # Hacky, would need to refactor JobBrowser get logs
- get_copy['format'] = 'python'
- request.GET = get_copy
- try:
- logs, workflow_action, is_really_done = self.get_log(request, job)
- progress = workflow_action[0]['progress']
- except:
- LOG.exception('failed to get progress')
- progress = 0
- else:
- progress = 100
- hue_pig = hue_jobs.get(job.id) and hue_jobs.get(job.id) or None
- massaged_job = {
- 'id': job.id,
- 'lastModTime': hasattr(job, 'lastModTime') and job.lastModTime and format_time(job.lastModTime) or None,
- 'kickoffTime': hasattr(job, 'kickoffTime') and job.kickoffTime or None,
- 'timeOut': hasattr(job, 'timeOut') and job.timeOut or None,
- 'endTime': job.endTime and format_time(job.endTime) or None,
- 'status': job.status,
- 'isRunning': job.is_running(),
- 'duration': job.endTime and job.startTime and format_duration_in_millis(( time.mktime(job.endTime) - time.mktime(job.startTime) ) * 1000) or None,
- 'appName': hue_pig and hue_pig.dict['name'] or _('Unsaved script'),
- 'scriptId': hue_pig and hue_pig.id or -1,
- 'scriptContent': hue_pig and hue_pig.dict['script'] or '',
- 'progress': progress,
- 'progressPercent': '%d%%' % progress,
- 'user': job.user,
- 'absoluteUrl': job.get_absolute_url(),
- 'canEdit': has_job_edition_permission(job, self.user),
- 'killUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'kill'}),
- 'watchUrl': reverse('pig:watch', kwargs={'job_id': job.id}) + '?format=python',
- 'created': hasattr(job, 'createdTime') and job.createdTime and job.createdTime and ((job.type == 'Bundle' and job.createdTime) or format_time(job.createdTime)),
- 'startTime': hasattr(job, 'startTime') and format_time(job.startTime) or None,
- 'run': hasattr(job, 'run') and job.run or 0,
- 'frequency': hasattr(job, 'frequency') and job.frequency or None,
- 'timeUnit': hasattr(job, 'timeUnit') and job.timeUnit or None,
- }
- jobs.append(massaged_job)
- return jobs
- def format_time(st_time):
- if st_time is None:
- return '-'
- else:
- return time.strftime("%a, %d %b %Y %H:%M:%S", st_time)
- def has_job_edition_permission(oozie_job, user):
- return user.is_superuser or oozie_job.user == user.username
|