瀏覽代碼

HUE-8509 [oozie] Schedule repetitive remote jobs

Romain Rigaux 7 年之前
父節點
當前提交
26d8d3b

+ 44 - 6
apps/oozie/src/oozie/models2.py

@@ -205,9 +205,11 @@ class WorkflowConfiguration(object):
     }
   ]
 
+
 class WorkflowDepthReached(Exception):
   pass
 
+
 class Workflow(Job):
   XML_FILE_NAME = 'workflow.xml'
   PROPERTY_APP_PATH = 'oozie.wf.application.path'
@@ -457,7 +459,8 @@ class Workflow(Job):
     node_mapping = dict([(node.id, node) for node in nodes])
     sub_wfs_ids = [node.data['properties']['workflow'] for node in nodes if node.data['type'] == 'subworkflow']
     workflow_mapping = dict(
-      [(workflow.uuid, Workflow(document=workflow, user=self.user)) for workflow in Document2.objects.filter(uuid__in=sub_wfs_ids)])
+        [(workflow.uuid, Workflow(document=workflow, user=self.user)) for workflow in Document2.objects.filter(uuid__in=sub_wfs_ids)]
+    )
 
     xml = re.sub(re.compile('>\s*\n+', re.MULTILINE), '>\n', django_mako.render_to_string(tmpl, {
       'wf': self,
@@ -554,6 +557,7 @@ def _to_lowercase(node_list):
       if hasattr(node[key], 'lower'):
         node[key] = node[key].lower()
 
+
 def _update_adj_list(adj_list):
   uuids = {}
   id = 1
@@ -589,6 +593,7 @@ def _update_adj_list(adj_list):
     id += 1
   return adj_list
 
+
 def _dig_nodes(nodes, adj_list, user, wf_nodes, nodes_uuid_set):
   for node in nodes:
     if type(node) != list:
@@ -659,6 +664,7 @@ def _dig_nodes(nodes, adj_list, user, wf_nodes, nodes_uuid_set):
     else:
       _dig_nodes(node, adj_list, user, wf_nodes, nodes_uuid_set)
 
+
 def _create_workflow_layout(nodes, adj_list, nodes_uuid_set, size=12):
   wf_rows = []
   for node in nodes:
@@ -666,11 +672,11 @@ def _create_workflow_layout(nodes, adj_list, nodes_uuid_set, size=12):
       node = node[0]
     if type(node) != list:
       _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id=adj_list[node]['uuid'],
-        row={"widgets":[{"size":size, "name": adj_list[node]['node_type'], "id":  adj_list[node]['uuid'], "widgetType": _get_widget_type(adj_list[node]['node_type']), "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
+        row = {"widgets":[{"size":size, "name": adj_list[node]['node_type'], "id":  adj_list[node]['uuid'], "widgetType": _get_widget_type(adj_list[node]['node_type']), "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
     else:
       if adj_list[node[0]]['node_type'] in ('fork', 'decision'):
         _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id=adj_list[node[0]]['uuid'],
-          row={"widgets":[{"size":size, "name": adj_list[node[0]]['name'], "id":  adj_list[node[0]]['uuid'], "widgetType": _get_widget_type(adj_list[node[0]]['node_type']), "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
+          row = {"widgets":[{"size":size, "name": adj_list[node[0]]['name'], "id":  adj_list[node[0]]['uuid'], "widgetType": _get_widget_type(adj_list[node[0]]['node_type']), "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
 
         wf_rows.append({
           "id": str(uuid.uuid4()),
@@ -703,12 +709,14 @@ def _get_widget_type(node_type):
   widget_name = "%s-widget" % node_type
   return widget_name if widget_name in NODES.keys() else 'generic-widget'
 
+
 # Prevent duplicate nodes in graph layout
 def _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id, row):
   if row['widgets'][0]['id'] not in nodes_uuid_set:
     nodes_uuid_set.add(row['widgets'][0]['id'])
     wf_rows.append(row)
 
+
 def _get_hierarchy_from_adj_list(adj_list, curr_node, node_hierarchy):
 
   _get_hierarchy_from_adj_list_helper(adj_list, curr_node, node_hierarchy, WORKFLOW_DEPTH_LIMIT)
@@ -767,6 +775,7 @@ def _create_graph_adjaceny_list(nodes):
 
 
 class Node():
+
   def __init__(self, data, user=None):
     self.data = data
     self.user = user
@@ -791,7 +800,9 @@ class Node():
                  % (len(links), len(self.data['children']), links, self.data['children']))
         self.data['children'] = links
 
-    if self.data['type'] == AltusAction.TYPE or ('altus' in mapping.get('cluster', '') and (self.data['type'] == SparkDocumentAction.TYPE or self.data['type'] == 'spark-document')):
+    if self.data['type'] == AltusAction.TYPE or \
+          (('altus' in mapping.get('cluster', '') and (self.data['type'] == SparkDocumentAction.TYPE or self.data['type'] == 'spark-document'))) or \
+          mapping.get('auto-cluster'):
       shell_command_name = self.data['name'] + '.sh'
       self.data['properties']['shell_command'] = shell_command_name
       self.data['properties']['env_var'] = []
@@ -930,7 +941,32 @@ class Node():
       'workflow_mapping': workflow_mapping
     }
 
-    if mapping.get('send_email'):
+    if mapping.get('auto-cluster'):
+      pass
+#       if self.data['type'] == StartNode.TYPE:
+#         self.data['altus_action'] = {
+#           'properties': {
+#             'credentials': {},
+#             'retry_max': {},
+#             'retry_interval': {},
+#             'prepares': {},
+#             'job_xml': {},
+#             'job_properties': {},
+#             'shell_command': '',
+#             'arguments': [],
+#             'env_var': [],
+#             'files': [],
+#             'archives': [],
+#             'capture_output': True
+#             #       <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
+# 
+#             #  Node(dict(AltusAction().get_fields()))
+#           }
+#         }
+#         self.data['properties']['auto-cluster'] = mapping['auto-cluster']
+#       if self.data['type'] == EndNode.TYPE or self.data['type'] == KillAction.TYPE:
+#         self.data['properties']['auto-cluster'] = mapping['auto-cluster']
+    elif mapping.get('send_email'):
       if self.data['type'] == KillAction.TYPE and not self.data['properties'].get('enableMail'):
         self.data['properties']['enableMail'] = True
         self.data['properties']['to'] = self.user.email
@@ -1000,7 +1036,9 @@ class Node():
       node_type = ShellAction.TYPE
     elif self.data['type'] == AltusAction.TYPE:
       node_type = ShellAction.TYPE
-    elif mapping.get('cluster') and 'document' in node_type:
+    elif mapping.get('cluster') and 'document' in node_type: # Workflow
+      node_type = ShellAction.TYPE
+    elif mapping.get('auto-cluster') and 'document' in node_type: # Scheduled workflow
       node_type = ShellAction.TYPE
 
     return 'editor2/gen/workflow-%s.xml.mako' % node_type

+ 39 - 0
apps/oozie/src/oozie/templates/editor2/gen/workflow-start.xml.mako

@@ -15,4 +15,43 @@
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 
+<%namespace name="common" file="workflow-common.xml.mako" />
+
+
+%if node['properties'].get('auto-cluster'):
+  <start to="${ node_mapping[node['children'][0]['to']].name }-start"/>
+
+  <action name="${ node['name'] }-start"${ common.credentials(node['altus_action']['properties']['credentials']) }${ common.retry_max(node['altus_action']['properties']['retry_max']) }${ common.retry_interval(node['altus_action']['properties']['retry_interval']) }>
+      <shell xmlns="uri:oozie:shell-action:0.1">
+          <job-tracker>${'${'}jobTracker}</job-tracker>
+          <name-node>${'${'}nameNode}</name-node>
+
+          ${ common.prepares(node['altus_action']['properties']['prepares']) }
+          % if node['altus_action']['properties']['job_xml']:
+            <job-xml>${ node['altus_action']['properties']['job_xml'] }</job-xml>
+          % endif
+          ${ common.configuration(node['altus_action']['properties']['job_properties']) }
+
+          <exec>${ node['altus_action']['properties']['shell_command'] }</exec>
+
+          % for param in node['altus_action']['properties']['arguments']:
+            <argument>${ param['value'] }</argument>
+          % endfor
+          
+          % for param in node['altus_action']['properties']['env_var']:
+            <env-var>${ param['value'] }</env-var>
+          % endfor            
+
+          ${ common.distributed_cache(node['altus_action']['properties']['files'], node['altus_action']['properties']['archives']) }
+
+          % if node['altus_action']['properties']['capture_output']:
+            <capture-output/>
+          % endif
+      </shell>
+      <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
+      ##<error to="${ node_mapping[node['children'][1]['error']].name }"/>
+      ${ common.sla(node) }
+  </action>
+%else:
     <start to="${ node_mapping[node['children'][0]['to']].name }"/>
+%endif

+ 51 - 0
apps/oozie/src/oozie/views/editor2.py

@@ -721,6 +721,57 @@ def submit_coordinator(request, doc_id):
 def _submit_coordinator(request, coordinator, mapping):
   try:
     wf = coordinator.workflow
+    mapping['auto-cluster'] = {
+      u'additionalClusterResourceTags': [],
+      u'automaticTerminationCondition': u'EMPTY_JOB_QUEUE', #'u'NONE',
+      u'cdhVersion': u'CDH514',
+      u'clouderaManagerPassword': u'guest',
+      u'clouderaManagerUsername': u'guest',
+      u'clusterName': u'analytics4', # Add time variable
+      u'computeWorkersConfiguration': {
+        u'bidUSDPerHr': 0,
+        u'groupSize': 0,
+        u'useSpot': False
+      },
+      u'environmentName': u'crn:altus:environments:us-west-1:12a0079b-1591-4ca0-b721-a446bda74e67:environment:analytics/236ebdda-18bd-428a-9d2b-cd6973d42946',
+      u'instanceBootstrapScript': u'',
+      u'instanceType': u'm4.xlarge',
+      u'jobSubmissionGroupName': u'',
+      u'jobs': [{
+          u'failureAction': u'INTERRUPT_JOB_QUEUE',
+          u'name': u'a87e20d7-5c0d-49ee-ab37-625fa2803d51',
+          u'sparkJob': {
+            u'applicationArguments': ['5'],
+            u'jars': [u's3a://datawarehouse-customer360/ETL/spark-examples.jar'],
+            u'mainClass': u'org.apache.spark.examples.SparkPi'
+          }
+        },
+#         {
+#           u'failureAction': u'INTERRUPT_JOB_QUEUE',
+#           u'name': u'a87e20d7-5c0d-49ee-ab37-625fa2803d51',
+#           u'sparkJob': {
+#             u'applicationArguments': ['10'],
+#             u'jars': [u's3a://datawarehouse-customer360/ETL/spark-examples.jar'],
+#             u'mainClass': u'org.apache.spark.examples.SparkPi'
+#           }
+#         },
+#         {
+#           u'failureAction': u'INTERRUPT_JOB_QUEUE',
+#           u'name': u'a87e20d7-5c0d-49ee-ab37-625fa2803d51',
+#           u'sparkJob': {
+#             u'applicationArguments': [u'filesystems3.conf'],
+#             u'jars': [u's3a://datawarehouse-customer360/ETL/envelope-0.6.0-SNAPSHOT-c6.jar'],
+#             u'mainClass': u'com.cloudera.labs.envelope.EnvelopeMain',
+#             u'sparkArguments': u'--archives=s3a://datawarehouse-customer360/ETL/filesystems3.conf'
+#           }
+#         }
+      ],
+      u'namespaceName': u'crn:altus:sdx:us-west-1:12a0079b-1591-4ca0-b721-a446bda74e67:namespace:analytics/7ea35fe5-dbc9-4b17-92b1-97a1ab32e410',
+      u'publicKey': u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDuTEfNIW8LEcVgprUrourbYjoW1RaTLhfzPnnBjJrg14koQrosl+s9phrpBBLTWmQuQdvy9iC2ma//gY5nz/7e+QuaeENhhoEiZn1PDBbFakD/AOjZXIu6DTEgCrOeXsQauFZKOkcFvrBGJC0qigYU3b8Eys4cun3RQ4S9WkDW6538wOSnsm6sXcL84KqbH+ay5gTk+lz3bi/6plALZMItbRz9IulXnLM4QfCwMxXTU/IjtnT+ltZVvKsWpfvDQ3Oyu/a6gK369iXcSP0e07KAzWiv2WYX46sNzZ8+de9ho1/VMaXnI4WrooV9lxByKWD+WsXkqtctT16VfxpX8CeR romain@unreal\\n',
+      u'serviceType': u'SPARK',
+      u'workersConfiguration': {},
+      u'workersGroupSize': u'3'
+    }
     wf_dir = Submission(request.user, wf, request.fs, request.jt, mapping, local_tz=coordinator.data['properties']['timezone']).deploy()
 
     properties = {'wf_application_path': request.fs.get_hdfs_path(wf_dir)}

+ 44 - 24
desktop/libs/liboozie/src/liboozie/submission2.py

@@ -210,8 +210,11 @@ class Submission(object):
           self.job.override_subworkflow_id(action, workflow.id) # For displaying the correct graph
           self.properties['workspace_%s' % workflow.uuid] = workspace # For pointing to the correct workspace
 
-        elif action.data['type'] == 'altus' or (action.data['type'] == 'spark-document' and 'altus' in self.properties.get('cluster', '')):
-          is_altus_job = 'altus' in self.properties.get('cluster', '')
+        elif action.data['type'] == 'altus' or \
+            (action.data['type'] == 'spark-document' and 'altus' in self.properties.get('cluster', '')) or \
+            (self.properties.get('auto-cluster') and 'document' in action.data['type']):
+          is_altus_job = 'altus' in self.properties.get('cluster', '') and action.data['type'] != 'altus'
+          is_scheduled_altus_job = self.properties.get('auto-cluster')
 
           self._create_file(deployment_dir, action.data['name'] + '.sh', '''#!/usr/bin/env bash
 
@@ -239,6 +242,14 @@ python altus.py
                 auth_key_id=ALTUS.AUTH_KEY_ID.get(),
                 auth_key_secret=ALTUS.AUTH_KEY_SECRET.get().replace('\\n', '\n')
             )
+          elif is_scheduled_altus_job:
+            shell_script = self._generate_altus_job_action_script(
+                service='dataeng',
+                cluster=self.properties['auto-cluster'],
+                jobs=[],
+                auth_key_id=ALTUS.AUTH_KEY_ID.get(),
+                auth_key_secret=ALTUS.AUTH_KEY_SECRET.get().replace('\\n', '\n')
+            )
           else:
             shell_script = self._generate_altus_action_script(
                 service=action.data['properties'].get('service'),
@@ -559,9 +570,7 @@ STORED AS TEXTFILE %s""" % (self.properties.get('send_result_path'), '\n\n\n'.jo
       if self._do_as(self.user.username , self.fs.exists, path):
         self._do_as(self.user.username , self.fs.rmtree, path)
     except Exception, ex:
-      LOG.warn("Failed to clean up workflow deployment directory for "
-               "%s (owner %s). Caused by: %s",
-               self.job.name, self.user, ex)
+      LOG.warn("Failed to clean up workflow deployment directory for %s (owner %s). Caused by: %s", self.job.name, self.user, ex)
 
   def _is_workflow(self):
     from oozie.models2 import Workflow
@@ -630,6 +639,13 @@ print _exec('%(service)s', '%(command)s', %(args)s)
     else:
       hostname = ALTUS.HOSTNAME.get()
 
+    if type(cluster) == dict:
+      command = 'createAWSCluster'
+      arguments = cluster
+    else:
+      command = 'submitJobs'
+      arguments = {'clusterName': cluster, 'jobs': jobs}
+
     return """#!/usr/bin/env python
 
 import time
@@ -639,7 +655,7 @@ from ast import literal_eval
 from navoptapi.api_lib import ApiLib
 
 hostname = '%(hostname)s'
-cluster = '%(cluster)s'
+arguments = literal_eval("%(arguments)s")
 auth_key_id = '%(auth_key_id)s'
 auth_key_secret = '''%(auth_key_secret)s'''
 
@@ -656,32 +672,36 @@ def _exec(service, command, parameters=None):
     raise e
 
 
-try:
-  handle = _exec('%(service)s', 'submitJobs', {'clusterName': cluster, 'jobs': literal_eval("%(jobs)s")})
+try:    
+  handle = _exec('%(service)s', '%(command)s', arguments)
   
-  job_id = handle['jobs'][0]['jobId']
-  status = 'QUEUED'
-  print 'Job submitted: %%s' %% job_id
-
-  while status in ('QUEUED', 'RUNNING', 'SUBMITTING'):
-    time.sleep(5)
+  if 'create' in '%(command)s':
+    handle = _exec('%(service)s', 'listJobs', {'clusterCrn': handle['cluster']['crn']})
 
-    print 'Checking status...'
-    status = _exec('%(service)s', 'describeJob', {'jobId': job_id})['job']['status']
-
-  if status != 'COMPLETED':
-    raise Exception('Job %%s failed %%s' %% (job_id, status))
-  else:
-    print 'Job %%s completed successfully' %% job_id
+  while handle['jobs']:
+    job = handle['jobs'].pop(0)
+    status = 'QUEUED'
+    print 'Job submitted: %%(jobId)s' %% job
+  
+    while status in ('QUEUED', 'RUNNING', 'SUBMITTING'):
+      time.sleep(5)
+  
+      print 'Checking status...'
+      status = _exec('%(service)s', 'describeJob', {'jobId': job['jobId']})['job']['status']
+  
+    if status != 'COMPLETED':
+      raise Exception('Job %%s failed %%s' %% (job['jobId'], status))
+    else:
+      print 'Job %%(jobId)s completed successfully' %% job
 except Exception, e:
   print e
   raise e
 
 """ % {
-      'hostname': hostname,
       'service': service,
-      'cluster': cluster,
-      'jobs': repr(jobs),
+      'hostname': hostname,      
+      'command': command,
+      'arguments': repr(arguments),
       'auth_key_id': auth_key_id,
       'auth_key_secret': auth_key_secret
     }

+ 0 - 5
desktop/libs/metadata/src/metadata/manager_api.py

@@ -21,11 +21,6 @@ import logging
 
 from metadata.manager_client import ManagerApi
 
-try:
-  from collections import OrderedDict
-except ImportError:
-  from ordereddict import OrderedDict # Python 2.6
-
 from django.http import Http404
 from django.utils.html import escape
 from django.utils.translation import ugettext as _

+ 1 - 1
desktop/libs/notebook/src/notebook/connectors/altus.py

@@ -217,7 +217,7 @@ class DataEngApi():
       u'serviceType': u'SPARK',
       u'workersConfiguration': {},
       u'workersGroupSize': u'3'
-  }
+    }
 
     return _exec('dataeng', 'createAWSCluster', params)