models.py 70 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. try:
  18. import json
  19. except ImportError:
  20. import simplejson as json
  21. import logging
  22. import re
  23. import time
  24. from datetime import datetime, timedelta
  25. from string import Template
  26. from itertools import chain
  27. from django.db import models
  28. from django.db.models import Q
  29. from django.core.urlresolvers import reverse
  30. from django.core.validators import RegexValidator
  31. from django.contrib.auth.models import User
  32. from django.contrib.contenttypes import generic
  33. from django.contrib.contenttypes.models import ContentType
  34. from django.forms.models import inlineformset_factory
  35. from django.utils.encoding import force_unicode
  36. from django.utils.translation import ugettext as _, ugettext_lazy as _t
  37. from desktop.log.access import access_warn
  38. from desktop.lib import django_mako
  39. from desktop.lib.exceptions_renderable import PopupException
  40. from desktop.lib.json_utils import JSONEncoderForHTML
  41. from desktop.models import Document
  42. from hadoop.fs.exceptions import WebHdfsException
  43. from hadoop.fs.hadoopfs import Hdfs
  44. from liboozie.submittion import Submission
  45. from liboozie.submittion import create_directories
  46. from oozie.conf import REMOTE_SAMPLE_DIR
  47. from timezones import TIMEZONES
  48. LOG = logging.getLogger(__name__)
  49. PATH_MAX = 512
  50. name_validator = RegexValidator(regex='^[a-zA-Z_][\-_a-zA-Z0-9]{1,39}$',
  51. message=_('Enter a valid value: combination of 2 - 40 letters and digits starting by a letter'))
  52. """
  53. Permissions:
  54. A Workflow/Coordinator can be accessed/submitted by its owner, a superuser or by anyone if its 'is_shared'
  55. property and SHARE_JOBS are set to True.
  56. A Workflow/Coordinator can be modified only by its owner or a superuser.
  57. Permissions checking happens by adding the decorators.
  58. """
  59. class JobManager(models.Manager):
  60. def can_read(self, user, job_id):
  61. job = Job.objects.select_related().get(pk=job_id).get_full_node()
  62. return job.can_read(user)
  63. def can_read_or_exception(self, request, job_id, exception_class=PopupException):
  64. if job_id is None:
  65. return
  66. try:
  67. job = Job.objects.select_related().get(pk=job_id).get_full_node()
  68. if job.can_read(request.user):
  69. return job
  70. else:
  71. message = _("Permission denied. %(username)s does not have the permissions required to access job %(id)s") % \
  72. {'username': request.user.username, 'id': job.id}
  73. access_warn(request, message)
  74. request.error(message)
  75. raise exception_class(message)
  76. except Job.DoesNotExist:
  77. raise exception_class(_('job %(id)s does not exist') % {'id': job_id})
  78. def can_edit_or_exception(self, request, job, exception_class=PopupException):
  79. if job.is_editable(request.user):
  80. return True
  81. else:
  82. raise exception_class(_('Not allowed to modified this job'))
  83. class Job(models.Model):
  84. """
  85. Base class for Oozie Workflows, Coordinators and Bundles.
  86. """
  87. owner = models.ForeignKey(User, db_index=True, verbose_name=_t('Owner'), help_text=_t('Person who can modify the job.')) # Deprecated
  88. name = models.CharField(max_length=40, blank=False, validators=[name_validator], # Deprecated
  89. help_text=_t('Name of the job, which must be unique per user.'), verbose_name=_t('Name'))
  90. description = models.CharField(max_length=1024, blank=True, verbose_name=_t('Description'), # Deprecated
  91. help_text=_t('The purpose of the job.'))
  92. last_modified = models.DateTimeField(auto_now=True, db_index=True, verbose_name=_t('Last modified'))
  93. schema_version = models.CharField(max_length=128, verbose_name=_t('Schema version'),
  94. help_text=_t('The version of the XML schema used to talk to Oozie.'))
  95. deployment_dir = models.CharField(max_length=1024, blank=True, verbose_name=_t('HDFS deployment directory'),
  96. help_text=_t('The path on the HDFS where all the workflows and '
  97. 'dependencies must be uploaded.'))
  98. is_shared = models.BooleanField(default=False, db_index=True, verbose_name=_t('Is shared'), # Deprecated
  99. help_text=_t('Enable other users to have access to this job.'))
  100. parameters = models.TextField(default='[{"name":"oozie.use.system.libpath","value":"true"}]', verbose_name=_t('Oozie parameters'),
  101. help_text=_t('Parameters used at the submission time (e.g. market=US, oozie.use.system.libpath=true).'))
  102. is_trashed = models.BooleanField(default=False, db_index=True, verbose_name=_t('Is trashed'), blank=True,# Deprecated
  103. help_text=_t('If this job is trashed.'))
  104. doc = generic.GenericRelation(Document, related_name='oozie_doc')
  105. objects = JobManager()
  106. unique_together = ('owner', 'name')
  107. def delete(self, skip_trash=False, *args, **kwargs):
  108. if skip_trash:
  109. self.doc.all().delete()
  110. return super(Job, self).delete(*args, **kwargs)
  111. else:
  112. for job in self.doc.all():
  113. job.send_to_trash()
  114. return self
  115. def restore(self):
  116. self.doc.get().restore_from_trash()
  117. return self
  118. def save(self):
  119. super(Job, self).save()
  120. if not self.deployment_dir:
  121. default_dir = Hdfs.join(REMOTE_SAMPLE_DIR.get(), '_%s_-oozie-%s-%s' % (self.owner.username, self.id, time.time()))
  122. self.deployment_dir = default_dir
  123. super(Job, self).save()
  124. def is_deployed(self, fs):
  125. return self.deployment_dir != '' and fs.exists(self.deployment_dir)
  126. def __str__(self):
  127. res = '%s - %s' % (force_unicode(self.name), self.owner)
  128. return force_unicode(res)
  129. def get_full_node(self):
  130. try:
  131. return self.workflow
  132. except Workflow.DoesNotExist:
  133. pass
  134. try:
  135. return self.coordinator
  136. except Coordinator.DoesNotExist:
  137. pass
  138. try:
  139. return self.bundle
  140. except Bundle.DoesNotExist:
  141. pass
  142. def get_type(self):
  143. return self.get_full_node().get_type()
  144. def get_absolute_url(self):
  145. return self.get_full_node().get_absolute_url()
  146. def get_parameters(self):
  147. return json.loads(self.parameters)
  148. @property
  149. def parameters_escapejs(self):
  150. return self._escapejs_parameters_list(self.parameters)
  151. def _escapejs_parameters_list(self, parameters):
  152. return json.dumps(json.loads(parameters), cls=JSONEncoderForHTML)
  153. @property
  154. def status(self):
  155. # TODO
  156. if self.is_shared:
  157. return _('shared')
  158. else:
  159. return _('personal')
  160. def find_all_parameters(self):
  161. params = self.find_parameters()
  162. for param in self.get_parameters():
  163. params[param['name'].strip()] = param['value']
  164. return [{'name': name, 'value': value} for name, value in params.iteritems()]
  165. def can_read(self, user):
  166. return self.doc.get().can_read(user)
  167. def is_editable(self, user):
  168. """Only owners or admins can modify a job."""
  169. return user.is_superuser or self.owner == user
  170. class WorkflowManager(models.Manager):
  171. def new_workflow(self, owner):
  172. workflow = Workflow(owner=owner, schema_version='uri:oozie:workflow:0.4')
  173. kill = Kill(name='kill', workflow=workflow, node_type=Kill.node_type)
  174. end = End(name='end', workflow=workflow, node_type=End.node_type)
  175. start = Start(name='start', workflow=workflow, node_type=Start.node_type)
  176. to = Link(parent=start, child=end, name='to')
  177. related = Link(parent=start, child=end, name='related')
  178. workflow.start = start
  179. workflow.end = end
  180. return workflow
  181. def initialize(self, workflow, fs):
  182. Kill.objects.create(name='kill', workflow=workflow, node_type=Kill.node_type)
  183. end = End.objects.create(name='end', workflow=workflow, node_type=End.node_type)
  184. start = Start.objects.create(name='start', workflow=workflow, node_type=Start.node_type)
  185. link = Link(parent=start, child=end, name='to')
  186. link.save()
  187. Link.objects.create(parent=start, child=end, name='related')
  188. workflow.start = start
  189. workflow.end = end
  190. workflow.save()
  191. Document.objects.link(workflow, owner=workflow.owner, name=workflow.name, description=workflow.description)
  192. self.check_workspace(workflow, fs)
  193. def check_workspace(self, workflow, fs):
  194. create_directories(fs, [REMOTE_SAMPLE_DIR.get()])
  195. create_directories(fs)
  196. if workflow.is_shared:
  197. perms = 0755
  198. else:
  199. perms = 0711
  200. Submission(workflow.owner, workflow, fs, {})._create_dir(workflow.deployment_dir, perms=perms)
  201. def destroy(self, workflow, fs):
  202. Submission(workflow.owner, workflow, fs, {}).remove_deployment_dir()
  203. try:
  204. workflow.coordinator_set.update(workflow=None) # In Django 1.3 could do ON DELETE set NULL
  205. except:
  206. pass
  207. workflow.save()
  208. workflow.delete(skip_trash=True)
  209. def managed(self):
  210. return self.filter(managed=True)
  211. def unmanaged(self):
  212. return self.filter(managed=False)
  213. class Workflow(Job):
  214. is_single = models.BooleanField(default=False)
  215. start = models.ForeignKey('Start', related_name='start_workflow', blank=True, null=True)
  216. end = models.ForeignKey('End', related_name='end_workflow', blank=True, null=True)
  217. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  218. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  219. 'Properties specified in the Job Properties element override properties specified in the '
  220. 'files specified in the Job XML element.'))
  221. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  222. help_text=_t('Job configuration properties used by all the actions of the workflow '
  223. '(e.g. mapred.job.queue.name=production)'))
  224. managed = models.BooleanField(default=True)
  225. objects = WorkflowManager()
  226. HUE_ID = 'hue-id-w'
  227. ICON = '/oozie/static/art/icon_oozie_24_workflow.png'
  228. def get_type(self):
  229. return 'workflow'
  230. def get_properties(self):
  231. return json.loads(self.job_properties)
  232. def clone(self, fs, new_owner=None):
  233. source_deployment_dir = self.deployment_dir # Needed
  234. nodes = self.node_set.all()
  235. copy_doc = self.doc.get().copy()
  236. links = Link.objects.filter(parent__workflow=self)
  237. copy = self
  238. copy.pk = None
  239. copy.id = None
  240. copy.name += '-copy'
  241. copy.deployment_dir = ''
  242. if new_owner is not None:
  243. copy.owner = new_owner
  244. copy.save()
  245. old_nodes_mapping = {}
  246. for node in nodes:
  247. prev_id = node.id
  248. node = node.get_full_node()
  249. node.pk = None
  250. node.id = None
  251. node.workflow = copy
  252. node.save()
  253. old_nodes_mapping[prev_id] = node
  254. for link in links:
  255. link.pk = None
  256. link.id = None
  257. link.parent = old_nodes_mapping[link.parent.id]
  258. link.child = old_nodes_mapping[link.child.id]
  259. link.save()
  260. copy.start = old_nodes_mapping[self.start.id]
  261. copy.end = old_nodes_mapping[self.end.id]
  262. copy.save()
  263. copy_doc.name = copy.name
  264. copy_doc.owner = copy.owner
  265. copy_doc.save()
  266. copy.doc.add(copy_doc)
  267. try:
  268. if copy.is_shared:
  269. perms = 0755
  270. else:
  271. perms = 0711
  272. fs.copy_remote_dir(source_deployment_dir, copy.deployment_dir, owner=copy.owner, dir_mode=perms)
  273. except WebHdfsException, e:
  274. msg = _('The copy of the deployment directory failed: %s.') % e
  275. LOG.error(msg)
  276. raise PopupException(msg)
  277. # Reload workflow from DB... clears relationship cache
  278. copy = Workflow.objects.get(id=copy.id)
  279. return copy
  280. @property
  281. def job_properties_escapejs(self):
  282. return self._escapejs_parameters_list(self.job_properties)
  283. def has_cycle(self):
  284. """
  285. Topological sort for detecting cycles in the directed graph.
  286. """
  287. queue = set([self.start])
  288. removed_edges = set()
  289. while queue:
  290. node = queue.pop()
  291. edges = set(node.get_children_links())
  292. for edge in edges:
  293. removed_edges.add(edge)
  294. # Edge has no other incoming edges
  295. if not set(edge.child.get_parent_links()) - removed_edges:
  296. queue.add(edge.child)
  297. graph_edges = set([edge for node in self.node_set.all() for edge in node.get_children_links()])
  298. return len(graph_edges - removed_edges) > 0 # Graph does not have unseen edges
  299. def find_parameters(self):
  300. params = set()
  301. for node in self.node_list:
  302. if hasattr(node, 'find_parameters'):
  303. params.update(node.find_parameters())
  304. return dict([(param, '') for param in list(params)])
  305. @property
  306. def actions(self):
  307. return Action.objects.filter(workflow=self, node_type__in=Action.types)
  308. @property
  309. def node_list(self):
  310. """Return a flatten node list ordered by the hierarchy of the nodes in the workflow"""
  311. def flatten(nodes):
  312. flat = []
  313. if type(nodes) == list:
  314. for node in nodes:
  315. flat.extend(flatten(node))
  316. else:
  317. flat.append(nodes)
  318. return flat
  319. def from_iterable(iterables):
  320. # Python 2.6 chain.from_iterable(['ABC', 'DEF']) --> A B C D E F
  321. for it in iterables:
  322. for element in it:
  323. yield element
  324. return list(chain(from_iterable([flatten(row) for row in self.get_hierarchy()])))
  325. @classmethod
  326. def get_application_path_key(cls):
  327. return 'oozie.wf.application.path'
  328. @classmethod
  329. def get_application_filename(cls):
  330. return 'workflow.xml'
  331. def get_absolute_url(self):
  332. if self.doc.get().extra == 'jobsub':
  333. return '/jobsub/#edit-design/%s' % self.id
  334. else:
  335. return reverse('oozie:edit_workflow', kwargs={'workflow': self.id}) + '#editWorkflow'
  336. def get_hierarchy(self):
  337. node = Start.objects.get(workflow=self) # Uncached version of start.
  338. kill = Kill.objects.get(workflow=node.workflow)
  339. # Special case: manage error email actions separately
  340. try:
  341. kill_nodes = [Link.objects.filter(child=kill).get(name='ok').parent, kill]
  342. except Link.DoesNotExist:
  343. kill_nodes = [kill]
  344. return self.get_hierarchy_rec(node=node) + [kill_nodes, [End.objects.get(workflow=node.workflow)]]
  345. def get_hierarchy_rec(self, node=None):
  346. if node is None:
  347. node = self.start
  348. if node.id is None:
  349. return []
  350. node = node.get_full_node()
  351. parents = node.get_parents()
  352. if isinstance(node, End):
  353. return [] # Not returning the end node
  354. elif isinstance(node, Decision):
  355. children = node.get_children('start')
  356. return [[node] + [[self.get_hierarchy_rec(node=child) for child in children],
  357. node.get_child_end()]] + self.get_hierarchy_rec(node.get_child_end().get_child('to'))
  358. elif isinstance(node, DecisionEnd):
  359. return []
  360. elif isinstance(node, Fork):
  361. children = node.get_children('start')
  362. return [[node] + [[self.get_hierarchy_rec(node=child) for child in children],
  363. node.get_child_join()]] + self.get_hierarchy_rec(node.get_child_join().get_child('to'))
  364. elif isinstance(node, Join):
  365. return []
  366. else:
  367. child = Link.objects.filter(parent=node).exclude(name__in=['related', 'kill', 'error'])[0].child
  368. return [node] + self.get_hierarchy_rec(child)
  369. def gen_status_graph(self, oozie_workflow):
  370. from oozie.forms import NodeMetaForm # Circular dependency
  371. actions = oozie_workflow.get_working_actions()
  372. controls = oozie_workflow.get_control_flow_actions()
  373. WorkflowFormSet = inlineformset_factory(Workflow, Node, form=NodeMetaForm, max_num=0, can_order=False, can_delete=False)
  374. forms = WorkflowFormSet(instance=self).forms
  375. template='editor/gen/workflow-graph-status.xml.mako'
  376. index = dict([(form.instance.id, form) for form in forms])
  377. actions_index = dict([(action.name, action) for action in actions])
  378. controls_index = dict([(control.name.strip(':'), control) for control in controls])
  379. return django_mako.render_to_string(template, {'nodes': self.get_hierarchy(), 'index': index, 'actions': actions_index, 'controls': controls_index})
  380. @classmethod
  381. def gen_status_graph_from_xml(cls, user, oozie_workflow):
  382. from oozie.import_workflow import import_workflow # Circular dependency
  383. try:
  384. workflow = Workflow.objects.new_workflow(user)
  385. workflow.save()
  386. try:
  387. import_workflow(workflow, oozie_workflow.definition)
  388. graph = workflow.gen_status_graph(oozie_workflow)
  389. return graph, workflow.node_list
  390. except Exception, e:
  391. LOG.warn('Workflow %s could not be converted to a graph: %s' % (oozie_workflow.id, e))
  392. finally:
  393. workflow.delete()
  394. return None, []
  395. def to_xml(self, mapping=None):
  396. if mapping is None:
  397. mapping = {}
  398. tmpl = 'editor/gen/workflow.xml.mako'
  399. xml = re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'workflow': self, 'mapping': mapping}))
  400. return force_unicode(xml)
  401. class Link(models.Model):
  402. # Links to exclude when using get_children_link(), get_parent_links() in the API
  403. META_LINKS = ('related',)
  404. parent = models.ForeignKey('Node', related_name='child_node')
  405. child = models.ForeignKey('Node', related_name='parent_node', verbose_name='')
  406. name = models.CharField(max_length=40)
  407. comment = models.CharField(max_length=1024, default='', blank=True)
  408. def __unicode__(self):
  409. return '%s %s %s' % (self.parent, self.child, self.name)
  410. class Node(models.Model):
  411. """
  412. Base class for the Oozie WorkflowAction or ControlFlow Nodes.
  413. http://nightly.cloudera.com/cdh4/cdh/4/oozie-3.1.3-cdh4.0.0-SNAPSHOT/WorkflowFunctionalSpec.html#a3_Workflow_Nodes
  414. The Node model is an abstract base class. All concrete actions derive from it.
  415. And it provides something for the Action or ControlFlow to reference.
  416. See https://docs.djangoproject.com/en/dev/topics/db/models/#multi-table-inheritance
  417. """
  418. PARAM_FIELDS = ()
  419. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name'),
  420. help_text=_t('Name of the action, which must be unique by workflow.'))
  421. description = models.CharField(max_length=1024, blank=True, default='', verbose_name=_t('Description'),
  422. help_text=_t('The purpose of the action.'))
  423. node_type = models.CharField(max_length=64, blank=False, verbose_name=_t('Type'),
  424. help_text=_t('The type of action (e.g. MapReduce, Pig...)'))
  425. workflow = models.ForeignKey(Workflow)
  426. children = models.ManyToManyField('self', related_name='parents', symmetrical=False, through=Link)
  427. unique_together = ('workflow', 'name')
  428. def get_full_node(self):
  429. if self.node_type == Mapreduce.node_type:
  430. node = self.mapreduce
  431. elif self.node_type == Pig.node_type:
  432. node = self.pig
  433. elif self.node_type == Hive.node_type:
  434. node = self.hive
  435. elif self.node_type == Sqoop.node_type:
  436. node = self.sqoop
  437. elif self.node_type == Ssh.node_type:
  438. node = self.ssh
  439. elif self.node_type == Shell.node_type:
  440. node = self.shell
  441. elif self.node_type == DistCp.node_type:
  442. node = self.distcp
  443. elif self.node_type == Fs.node_type:
  444. node = self.fs
  445. elif self.node_type == Email.node_type:
  446. node = self.email
  447. elif self.node_type == SubWorkflow.node_type:
  448. node = self.subworkflow
  449. elif self.node_type == Streaming.node_type:
  450. node = self.streaming
  451. elif self.node_type == Java.node_type:
  452. node = self.java
  453. elif self.node_type == Generic.node_type:
  454. node = self.generic
  455. elif self.node_type == Start.node_type:
  456. node = self.start
  457. elif self.node_type == End.node_type:
  458. node = self.end
  459. elif self.node_type == Kill.node_type:
  460. node = self.kill
  461. elif self.node_type == Fork.node_type:
  462. node = self.fork
  463. elif self.node_type == Join.node_type:
  464. node = self.join
  465. elif self.node_type == Decision.node_type:
  466. node = self.decision
  467. elif self.node_type == DecisionEnd.node_type:
  468. node = self.decisionend
  469. else:
  470. raise Exception(_('Unknown Node type: %s. Was it set at its creation?'), (self.node_type,))
  471. return node
  472. def find_parameters(self):
  473. return find_parameters(self, self.PARAM_FIELDS)
  474. def __unicode__(self):
  475. if self.name != '':
  476. return '%s' % self.name
  477. else:
  478. return '%s-%s' % (self.node_type, self.id)
  479. def to_xml(self, mapping=None):
  480. if mapping is None:
  481. mapping = {}
  482. node = self.get_full_node()
  483. data = {
  484. 'node': node,
  485. 'mapping': mapping
  486. }
  487. return django_mako.render_to_string(node.get_template_name(), data)
  488. # Can't use through relation directly with this Django version?
  489. # https://docs.djangoproject.com/en/1.2/topics/db/models/#intermediary-manytomany
  490. def get_link(self, name=None):
  491. if name is None:
  492. return Link.objects.exclude(name__in=Link.META_LINKS).get(parent=self)
  493. else:
  494. return Link.objects.exclude(name__in=Link.META_LINKS).get(parent=self, name=name)
  495. def get_child_link(self, name=None):
  496. return self.get_link(name)
  497. def get_child(self, name=None):
  498. """Includes DecisionEnd nodes"""
  499. return self.get_link(name).child.get_full_node()
  500. def get_oozie_child(self, name=None):
  501. """Resolves DecisionEnd nodes"""
  502. child = self.get_link(name).child.get_full_node()
  503. if child and child.node_type == DecisionEnd.node_type:
  504. child = child.get_oozie_child('to')
  505. return child
  506. def get_children(self, name=None):
  507. if name is not None:
  508. return [link.child for link in Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self, name=name)]
  509. else:
  510. return [link.child for link in Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self)]
  511. def get_parent(self, name=None):
  512. if name is not None:
  513. return self.get_parent_link(name).parent.get_full_node()
  514. else:
  515. return self.get_parent_link().parent.get_full_node()
  516. def get_parents(self):
  517. return [link.parent for link in self.get_parent_links()]
  518. def get_parent_link(self, name=None):
  519. if name is not None:
  520. return Link.objects.get(child=self, name=name)
  521. else:
  522. return Link.objects.get(child=self)
  523. def get_parent_links(self):
  524. return Link.objects.filter(child=self).exclude(name__in=Link.META_LINKS)
  525. def get_children_links(self, name=None):
  526. if name is None:
  527. return Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self)
  528. else:
  529. return Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self, name=name)
  530. def get_all_children_links(self):
  531. return Link.objects.filter(parent=self)
  532. def get_template_name(self):
  533. return 'editor/gen/workflow-%s.xml.mako' % self.node_type
  534. def is_visible(self):
  535. return True
  536. def add_node(self, child):
  537. raise NotImplementedError(_("%(node_type)s has not implemented the 'add_node' method.") % {
  538. 'node_type': self.node_type
  539. })
  540. class Action(Node):
  541. types = ()
  542. class Meta:
  543. # Cloning does not work anymore if not abstract
  544. abstract = True
  545. def add_node(self, child):
  546. Link.objects.filter(parent=self, name='ok').delete()
  547. Link.objects.create(parent=self, child=child, name='ok')
  548. if not Link.objects.filter(parent=self, name='error').exists():
  549. Link.objects.create(parent=self, child=Kill.objects.get(name='kill', workflow=self.workflow), name='error')
  550. # The fields with '[]' as default value are JSON dictionaries
  551. # When adding a new action, also update
  552. # - Action.types below
  553. # - Node.get_full_node()
  554. # - forms.py _node_type_TO_FORM_CLS
  555. # - workflow.js
  556. # - maybe actions_utils.mako
  557. class Mapreduce(Action):
  558. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'jar_path', 'prepares')
  559. node_type = 'mapreduce'
  560. files = models.TextField(default="[]", verbose_name=_t('Files'),
  561. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  562. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  563. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  564. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  565. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  566. jar_path = models.CharField(max_length=PATH_MAX, verbose_name=_t('Jar name'),
  567. help_text=_t('Name or path to the %(program)s jar file on HDFS. E.g. examples.jar.') % {'program': 'MapReduce'})
  568. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  569. help_text=_t('List of absolute paths to delete and then to create before starting the application. '
  570. 'This should be used exclusively for directory cleanup.'))
  571. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  572. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  573. 'Properties specified in the Job Properties element override properties specified in the '
  574. 'files specified in the Job XML element.'))
  575. def get_properties(self):
  576. return json.loads(self.job_properties)
  577. def get_files(self):
  578. return json.loads(self.files)
  579. def get_archives(self):
  580. return json.loads(self.archives)
  581. def get_prepares(self):
  582. return json.loads(self.prepares)
  583. class Streaming(Action):
  584. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'mapper', 'reducer')
  585. node_type = "streaming"
  586. files = models.TextField(default="[]", verbose_name=_t('Files'),
  587. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  588. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  589. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  590. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  591. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  592. mapper = models.CharField(max_length=PATH_MAX, blank=False, verbose_name=_t('Mapper'),
  593. help_text=_t('The executable/script to be used as mapper.'))
  594. reducer = models.CharField(max_length=PATH_MAX, blank=False, verbose_name=_t('Reducer'),
  595. help_text=_t('The executable/script to be used as reducer.'))
  596. def get_properties(self):
  597. return json.loads(self.job_properties)
  598. def get_files(self):
  599. return json.loads(self.files)
  600. def get_archives(self):
  601. return json.loads(self.archives)
  602. class Java(Action):
  603. PARAM_FIELDS = ('files', 'archives', 'jar_path', 'main_class', 'args',
  604. 'java_opts', 'job_properties', 'prepares')
  605. node_type = "java"
  606. files = models.TextField(default="[]", verbose_name=_t('Files'),
  607. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  608. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  609. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  610. jar_path = models.CharField(max_length=PATH_MAX, blank=False, verbose_name=_t('Jar name'),
  611. help_text=_t('Name or path to the %(program)s jar file on HDFS. E.g. examples.jar.') % {'program': 'Java'})
  612. main_class = models.CharField(max_length=256, blank=False, verbose_name=_t('Main class'),
  613. help_text=_t('Full name of the Java class. E.g. org.apache.hadoop.examples.Grep'))
  614. args = models.TextField(blank=True, verbose_name=_t('Arguments'),
  615. help_text=_t('Arguments of the main method. The value of each arg element is considered a single argument '
  616. 'and they are passed to the main method in the same order.'))
  617. java_opts = models.CharField(max_length=256, blank=True, verbose_name=_t('Java options'),
  618. help_text=_t('Command-line parameters used to start the JVM that will execute '
  619. 'the Java application. Using this element is equivalent to using the mapred.child.java.opts '
  620. 'configuration property. E.g. -Dexample-property=hue'))
  621. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  622. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  623. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  624. help_text=_t('List of absolute paths to delete and then to create before starting the application. '
  625. 'This should be used exclusively for directory cleanup.'))
  626. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  627. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  628. 'Properties specified in the Job Properties element override properties specified in the '
  629. 'files specified in the Job XML element.'))
  630. capture_output = models.BooleanField(default=False, verbose_name=_t('Capture output'),
  631. help_text=_t('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  632. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  633. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  634. 'via the String action:output(String node, String key) function') % {'program': node_type.title()})
  635. def get_properties(self):
  636. return json.loads(self.job_properties)
  637. def get_files(self):
  638. return json.loads(self.files)
  639. def get_archives(self):
  640. return json.loads(self.archives)
  641. def get_prepares(self):
  642. return json.loads(self.prepares)
  643. class Pig(Action):
  644. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  645. node_type = 'pig'
  646. script_path = models.CharField(max_length=256, blank=False, verbose_name=_t('Script name'),
  647. help_text=_t('Script name or path to the Pig script. E.g. my_script.pig.'))
  648. params = models.TextField(default="[]", verbose_name=_t('Parameters'),
  649. help_text=_t('The Pig parameters of the script. e.g. "-param", "INPUT=${inputDir}"'))
  650. files = models.TextField(default="[]", verbose_name=_t('Files'),
  651. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  652. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  653. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  654. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  655. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  656. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  657. help_text=_t('List of absolute paths to delete and then to create before starting the application. '
  658. 'This should be used exclusively for directory cleanup.'))
  659. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  660. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  661. 'Properties specified in the Job Properties element override properties specified in the '
  662. 'files specified in the Job XML element.'))
  663. def get_properties(self):
  664. return json.loads(self.job_properties)
  665. def get_files(self):
  666. return json.loads(self.files)
  667. def get_archives(self):
  668. return json.loads(self.archives)
  669. def get_params(self):
  670. return json.loads(self.params)
  671. def get_prepares(self):
  672. return json.loads(self.prepares)
  673. class Hive(Action):
  674. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  675. node_type = 'hive'
  676. script_path = models.CharField(max_length=256, blank=False, verbose_name=_t('Script name'),
  677. help_text=_t('Script name or path to the %(type)s script. E.g. my_script.sql.') % {'type': node_type.title()})
  678. params = models.TextField(default="[]", verbose_name=_t('Parameters'),
  679. help_text=_t('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': node_type.title()})
  680. files = models.TextField(default="[]", verbose_name=_t('Files'),
  681. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  682. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  683. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  684. job_properties = models.TextField(default='[{"name":"oozie.hive.defaults","value":"hive-site.xml"}]',
  685. verbose_name=_t('Hadoop job properties'),
  686. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  687. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  688. help_text=_t('List of absolute paths to delete, then create, before starting the application. '
  689. 'This should be used exclusively for directory cleanup.'))
  690. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  691. help_text=_t('Refer to a Hive hive-site.xml file bundled in the workflow deployment directory. '))
  692. def get_properties(self):
  693. return json.loads(self.job_properties)
  694. def get_files(self):
  695. return json.loads(self.files)
  696. def get_archives(self):
  697. return json.loads(self.archives)
  698. def get_params(self):
  699. return json.loads(self.params)
  700. def get_prepares(self):
  701. return json.loads(self.prepares)
  702. class Sqoop(Action):
  703. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  704. node_type = 'sqoop'
  705. script_path = models.TextField(blank=True, verbose_name=_t('Command'), default='',
  706. help_text=_t('The full %(type)s command. Either put it here or split it by spaces and insert the parts as multiple parameters below.')
  707. % {'type': node_type.title()})
  708. params = models.TextField(default="[]", verbose_name=_t('Parameters'),
  709. help_text=_t('If no command is specified, split the command by spaces and insert the %(type)s parameters '
  710. 'here e.g. import, --connect, jdbc:hsqldb:file:db.hsqldb, ...') % {'type': node_type.title()})
  711. files = models.TextField(default="[]", verbose_name=_t('Files'),
  712. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  713. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  714. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  715. job_properties = models.TextField(default='[]',
  716. verbose_name=_t('Hadoop job properties'),
  717. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  718. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  719. help_text=_t('List of absolute paths to delete then to create before starting the application. '
  720. 'This should be used exclusively for directory cleanup'))
  721. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  722. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  723. 'Properties specified in the Job Properties element override properties specified in the '
  724. 'files specified in the Job XML element.'))
  725. def get_properties(self):
  726. return json.loads(self.job_properties)
  727. def get_files(self):
  728. return json.loads(self.files)
  729. def get_archives(self):
  730. return json.loads(self.archives)
  731. def get_params(self):
  732. return json.loads(self.params)
  733. def get_prepares(self):
  734. return json.loads(self.prepares)
  735. class Ssh(Action):
  736. PARAM_FIELDS = ('user', 'host', 'command', 'params')
  737. node_type = 'ssh'
  738. user = models.CharField(max_length=64, verbose_name=_t('User'),
  739. help_text=_t('User executing the shell command.'))
  740. host = models.CharField(max_length=256, verbose_name=_t('Host'),
  741. help_text=_t('Where the shell will be executed.'))
  742. command = models.CharField(max_length=256, verbose_name=_t('%(type)s command') % {'type': node_type.title()},
  743. help_text=_t('The command that will be executed.'))
  744. params = models.TextField(default="[]", verbose_name=_t('Arguments'),
  745. help_text=_t('The arguments of the %(type)s command.') % {'type': node_type.title()})
  746. capture_output = models.BooleanField(default=False, verbose_name=_t('Capture output'),
  747. help_text=_t('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  748. 'command output must be in Java properties file format and it must not exceed 2KB. '
  749. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  750. 'via the String action:output(String node, String key) function') % {'program': node_type.title()})
  751. def get_params(self):
  752. return json.loads(self.params)
  753. class Shell(Action):
  754. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  755. node_type = 'shell'
  756. command = models.CharField(max_length=256, blank=False, verbose_name=_t('%(type)s command') % {'type': node_type.title()},
  757. help_text=_t('The path of the Shell command to execute.'))
  758. params = models.TextField(default="[]", verbose_name=_t('Arguments'),
  759. help_text=_t('The arguments of Shell command can then be specified using one or more argument element.'))
  760. files = models.TextField(default="[]", verbose_name=_t('Files'),
  761. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  762. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  763. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  764. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  765. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  766. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  767. help_text=_t('List of absolute paths to delete then to create before starting the application. '
  768. 'This should be used exclusively for directory cleanup'))
  769. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  770. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  771. 'Properties specified in the Job Properties element override properties specified in the '
  772. 'files specified in the Job XML element.'))
  773. capture_output = models.BooleanField(default=False, verbose_name=_t('Capture output'),
  774. help_text=_t('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  775. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  776. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  777. 'via the String action:output(String node, String key) function') % {'program': node_type.title()})
  778. def get_properties(self):
  779. return json.loads(self.job_properties)
  780. def get_files(self):
  781. return json.loads(self.files)
  782. def get_archives(self):
  783. return json.loads(self.archives)
  784. def get_params(self):
  785. return json.loads(self.params)
  786. def get_prepares(self):
  787. return json.loads(self.prepares)
  788. class DistCp(Action):
  789. PARAM_FIELDS = ('job_properties', 'params', 'prepares')
  790. node_type = 'distcp'
  791. params = models.TextField(default="[]", verbose_name=_t('Arguments'),
  792. help_text=_t('The arguments of the %(type)s command. Put options first, then source paths, then destination path.')
  793. % {'type': node_type.title()})
  794. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  795. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production'))
  796. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  797. help_text=_t('List of absolute paths to delete then to create before starting the application. '
  798. 'This should be used exclusively for directory cleanup'))
  799. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  800. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  801. 'Properties specified in the Job Properties element override properties specified in the '
  802. 'files specified in the Job XML element.'))
  803. def get_properties(self):
  804. return json.loads(self.job_properties)
  805. def get_params(self):
  806. return json.loads(self.params)
  807. def get_prepares(self):
  808. return json.loads(self.prepares)
  809. class Fs(Action):
  810. PARAM_FIELDS = ('deletes', 'mkdirs', 'moves', 'chmods', 'touchzs')
  811. node_type = 'fs'
  812. deletes = models.TextField(default="[]", verbose_name=_t('Delete path'), blank=True,
  813. help_text=_t('Delete the specified path, if it is a directory it deletes recursively all its content and '
  814. 'then deletes the directory.'))
  815. mkdirs = models.TextField(default="[]", verbose_name=_t('Create directory'), blank=True,
  816. help_text=_t('Create the specified directory, it creates all missing directories in the path. '
  817. 'If the directory already exist it does a no-op.'))
  818. moves = models.TextField(default="[]", verbose_name=_t('Move file'), blank=True,
  819. help_text=_t('Move a file or directory to another path.'))
  820. chmods = models.TextField(default="[]", verbose_name=_t('Change permissions'), blank=True,
  821. help_text=_t('Change the permissions for the specified path. Permissions can be specified using the Unix Symbolic '
  822. 'representation (e.g. -rwxrw-rw-) or an octal representation (755).'))
  823. touchzs = models.TextField(default="[]", verbose_name=_t('Create or touch a file'), blank=True,
  824. help_text=_t('Creates a zero length file in the specified path if none exists or touch it.'))
  825. def get_deletes(self):
  826. return json.loads(self.deletes)
  827. def get_mkdirs(self):
  828. return json.loads(self.mkdirs)
  829. def get_moves(self):
  830. return json.loads(self.moves)
  831. def get_chmods(self):
  832. return json.loads(self.chmods)
  833. def get_touchzs(self):
  834. return json.loads(self.touchzs)
  835. class Email(Action):
  836. PARAM_FIELDS = ('to', 'cc', 'subject', 'body')
  837. node_type = 'email'
  838. to = models.TextField(default='', verbose_name=_t('TO addresses'), help_text=_t('Comma-separated values.'))
  839. cc = models.TextField(default='', verbose_name=_t('CC addresses (optional)'), blank=True, help_text=_t('Comma-separated values.'))
  840. subject = models.TextField(default='', verbose_name=_t('Subject'), help_text=_t('Plain-text.'))
  841. body = models.TextField(default='', verbose_name=_t('Body'), help_text=_t('Plain-text.'))
  842. class SubWorkflow(Action):
  843. PARAM_FIELDS = ('subworkflow', 'propagate_configuration', 'job_properties')
  844. node_type = 'subworkflow'
  845. sub_workflow = models.ForeignKey(Workflow, db_index=True, verbose_name=_t('Sub-workflow'),
  846. help_text=_t('The sub-workflow application to include. You must own all the sub-workflows.'))
  847. propagate_configuration = models.BooleanField(default=True, verbose_name=_t('Propagate configuration'), blank=True,
  848. help_text=_t('If the workflow job configuration should be propagated to the child workflow.'))
  849. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  850. help_text=_t('Can be used to specify the job properties that are required to run the child workflow job.'))
  851. def get_properties(self):
  852. return json.loads(self.job_properties)
  853. class Generic(Action):
  854. PARAM_FIELDS = ('xml',)
  855. node_type = 'generic'
  856. xml = models.TextField(default='', verbose_name=_t('XML of the custom action'),
  857. help_text=_t('This will be inserted verbatim in the action %(action)s. '
  858. 'E.g. all the XML content like %(xml_action)s '
  859. 'will be inserted into the action and produce %(full_action)s') % {
  860. 'action': '<action name="email">...</action>',
  861. 'xml_action': '<email><cc>hue@hue.org</cc></email>',
  862. 'full_action': '<action name="email"><email><cc>hue@hue.org</cc></email><ok/><error/></action>'})
  863. Action.types = (Mapreduce.node_type, Streaming.node_type, Java.node_type, Pig.node_type, Hive.node_type, Sqoop.node_type, Ssh.node_type, Shell.node_type,
  864. DistCp.node_type, Fs.node_type, Email.node_type, SubWorkflow.node_type, Generic.node_type)
  865. class ControlFlow(Node):
  866. """
  867. http://incubator.apache.org/oozie/docs/3.2.0-incubating/docs/WorkflowFunctionalSpec.html#a3.1_Control_Flow_Nodes
  868. """
  869. class Meta:
  870. abstract = True
  871. def get_xml(self):
  872. return django_mako.render_to_string(self.get_template_name(), {})
  873. def is_visible(self):
  874. return True
  875. # Could not make this abstract
  876. class Start(ControlFlow):
  877. node_type = 'start'
  878. def add_node(self, child):
  879. Link.objects.filter(parent=self).delete()
  880. link = Link.objects.create(parent=self, child=child, name='to')
  881. class End(ControlFlow):
  882. node_type = 'end'
  883. def add_node(self, child):
  884. raise RuntimeError(_("End should not have any children."))
  885. class Kill(ControlFlow):
  886. node_type = 'kill'
  887. message = models.CharField(max_length=256, blank=False, default='Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]')
  888. def add_node(self, child):
  889. raise RuntimeError(_("Kill should not have any children."))
  890. def is_visible(self):
  891. return False
  892. class Fork(ControlFlow):
  893. """
  894. A Fork can be converted into a Decision node.
  895. """
  896. node_type = 'fork'
  897. def is_visible(self):
  898. return True
  899. def get_child_join(self):
  900. return Link.objects.get(parent=self, name='related').child.get_full_node()
  901. def convert_to_decision(self):
  902. self.remove_join()
  903. decision = Decision.objects.create(workflow=self.workflow, node_type=Decision.node_type)
  904. decision.save()
  905. links = self.get_all_children_links()
  906. has_default = False
  907. for link in links:
  908. if link.name == 'default':
  909. has_default = True
  910. link.parent = decision
  911. # Defaults to end
  912. if not has_default:
  913. link = Link.objects.create(name="default", parent=decision, child=self.workflow.end)
  914. link.save()
  915. self.delete()
  916. return decision
  917. def remove_join(self):
  918. join = self.get_child_join()
  919. after_join = join.get_child('to')
  920. for parent in join.get_parent_actions():
  921. link = parent.get_link('ok')
  922. link.child = after_join
  923. link.save()
  924. # Automatically delete links thought foreign keys
  925. join.delete()
  926. class Join(ControlFlow):
  927. node_type = 'join'
  928. def is_visible(self):
  929. return True
  930. def get_parent_fork(self):
  931. return self.get_parent_link('related').parent.get_full_node()
  932. def get_parent_actions(self):
  933. return [link.parent for link in self.get_parent_links()]
  934. class Decision(ControlFlow):
  935. """
  936. Essentially a fork where only one of the paths of execution are chosen.
  937. Graphically, this is represented the same way as a fork.
  938. The DecisionEnd node is not represented in Oozie, only in Hue.
  939. """
  940. node_type = 'decision'
  941. def get_child_end(self):
  942. return Link.objects.get(parent=self, name='related').child.get_full_node()
  943. def is_visible(self):
  944. return True
  945. def update_description(self):
  946. self.description = ', '.join(self.get_children_links().values_list('comment', flat=True))
  947. self.save()
  948. class DecisionEnd(ControlFlow):
  949. """
  950. Defines the end of a join.
  951. This node exists purely in the Hue application to provide a smooth transition
  952. from Decision to Endself.
  953. NOTE: NOT AN OOZIE NODE
  954. """
  955. node_type = 'decisionend'
  956. def is_visible(self):
  957. return False
  958. def get_parent_decision(self):
  959. return self.get_parent_link('related').parent.get_full_node()
  960. def get_parent_actions(self):
  961. return [link.parent for link in self.get_parent_links()]
  962. def to_xml(self, mapping):
  963. return ''
  964. FREQUENCY_UNITS = (('minutes', _('Minutes')),
  965. ('hours', _('Hours')),
  966. ('days', _('Days')),
  967. ('months', _('Months')))
  968. FREQUENCY_NUMBERS = [(i, i) for i in xrange(1, 61)]
  969. DATASET_FREQUENCY = ['MINUTE', 'HOUR', 'DAY', 'MONTH', 'YEAR']
  970. class Coordinator(Job):
  971. frequency_number = models.SmallIntegerField(default=1, choices=FREQUENCY_NUMBERS, verbose_name=_t('Frequency number'),
  972. help_text=_t('The number of units of the rate at which '
  973. 'data is periodically created.'))
  974. frequency_unit = models.CharField(max_length=20, choices=FREQUENCY_UNITS, default='days', verbose_name=_t('Frequency unit'),
  975. help_text=_t('The unit of the rate at which data is periodically created.'))
  976. timezone = models.CharField(max_length=24, choices=TIMEZONES, default='America/Los_Angeles', verbose_name=_t('Timezone'),
  977. help_text=_t('The timezone of the coordinator.'))
  978. start = models.DateTimeField(default=datetime.today(), verbose_name=_t('Start'),
  979. help_text=_t('When to start the first workflow.'))
  980. end = models.DateTimeField(default=datetime.today() + timedelta(days=3), verbose_name=_t('End'),
  981. help_text=_t('When to start the last workflow.'))
  982. workflow = models.ForeignKey(Workflow, null=True, verbose_name=_t('Workflow'),
  983. help_text=_t('The workflow to schedule repeatedly.'))
  984. timeout = models.SmallIntegerField(null=True, blank=True, verbose_name=_t('Timeout'),
  985. help_text=_t('Number of minutes the coordinator action will be in '
  986. 'WAITING or READY status before giving up on its execution.'))
  987. concurrency = models.PositiveSmallIntegerField(null=True, blank=True, choices=FREQUENCY_NUMBERS, verbose_name=_t('Concurrency'),
  988. help_text=_t('The number of coordinator actions that are allowed to run concurrently (RUNNING status) '
  989. 'before the coordinator engine starts throttling them.'))
  990. execution = models.CharField(max_length=10, null=True, blank=True, verbose_name=_t('Execution'),
  991. choices=(('FIFO', _t('FIFO (oldest first) default')),
  992. ('LIFO', _t('LIFO (newest first)')),
  993. ('LAST_ONLY', _t('LAST_ONLY (discards all older materializations)'))),
  994. help_text=_t('Execution strategy of its coordinator actions when there is backlog of coordinator '
  995. 'actions in the coordinator engine. The different execution strategies are \'oldest first\', '
  996. '\'newest first\' and \'last one only\'. A backlog normally happens because of delayed '
  997. 'input data, concurrency control or because manual re-runs of coordinator jobs.'))
  998. throttle = models.PositiveSmallIntegerField(null=True, blank=True, choices=FREQUENCY_NUMBERS, verbose_name=_t('Throttle'),
  999. help_text=_t('The materialization or creation throttle value for its coordinator actions. '
  1000. 'Number of maximum coordinator actions that are allowed to be in WAITING state concurrently.'))
  1001. job_properties = models.TextField(default='[]', verbose_name=_t('Workflow properties'),
  1002. help_text=_t('Additional properties to transmit to the workflow, e.g. limit=100, and EL functions, e.g. username=${coord:user()}'))
  1003. HUE_ID = 'hue-id-c'
  1004. ICON = '/oozie/static/art/icon_oozie_24_coordinator.png'
  1005. def get_type(self):
  1006. return 'coordinator'
  1007. def to_xml(self, mapping=None):
  1008. if mapping is None:
  1009. mapping = {}
  1010. tmpl = "editor/gen/coordinator.xml.mako"
  1011. return re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'coord': self, 'mapping': mapping})).encode('utf-8', 'xmlcharrefreplace')
  1012. def clone(self, new_owner=None):
  1013. copy_doc = self.doc.get()
  1014. datasets = Dataset.objects.filter(coordinator=self)
  1015. data_inputs = DataInput.objects.filter(coordinator=self)
  1016. data_outputs = DataOutput.objects.filter(coordinator=self)
  1017. copy = self
  1018. copy.pk = None
  1019. copy.id = None
  1020. copy.name += '-copy'
  1021. copy.deployment_dir = ''
  1022. if new_owner is not None:
  1023. copy.owner = new_owner
  1024. copy.save()
  1025. old_dataset_mapping = {}
  1026. for dataset in datasets:
  1027. prev_id = dataset.id
  1028. dataset.pk = None
  1029. dataset.id = None
  1030. dataset.coordinator = copy
  1031. dataset.save()
  1032. old_dataset_mapping[prev_id] = dataset
  1033. for data_input in data_inputs:
  1034. data_input.pk = None
  1035. data_input.id = None
  1036. data_input.coordinator = copy
  1037. data_input.dataset = old_dataset_mapping[data_input.dataset.id]
  1038. data_input.save()
  1039. for data_output in data_outputs:
  1040. data_output.pk = None
  1041. data_output.id = None
  1042. data_output.coordinator = copy
  1043. data_output.dataset = old_dataset_mapping[data_output.dataset.id]
  1044. data_output.save()
  1045. copy_doc.pk = None
  1046. copy_doc.id = None
  1047. copy_doc.owner = copy.owner
  1048. copy_doc.save()
  1049. copy.doc.add(copy_doc)
  1050. return copy
  1051. @classmethod
  1052. def get_application_path_key(cls):
  1053. return 'oozie.coord.application.path'
  1054. @classmethod
  1055. def get_application_filename(cls):
  1056. return 'coordinator.xml'
  1057. def get_properties(self):
  1058. props = json.loads(self.job_properties)
  1059. index = [prop['name'] for prop in props]
  1060. for prop in self.workflow.get_parameters():
  1061. if not prop['name'] in index:
  1062. props.append(prop)
  1063. return props
  1064. @property
  1065. def job_properties_escapejs(self):
  1066. return self._escapejs_parameters_list(self.job_properties)
  1067. @property
  1068. def start_utc(self):
  1069. return utc_datetime_format(self.start)
  1070. @property
  1071. def end_utc(self):
  1072. return utc_datetime_format(self.end)
  1073. def get_absolute_url(self):
  1074. return reverse('oozie:edit_coordinator', kwargs={'coordinator': self.id})
  1075. @property
  1076. def frequency(self):
  1077. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1078. @property
  1079. def text_frequency(self):
  1080. return '%(number)d %(unit)s' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1081. def find_parameters(self):
  1082. params = self.workflow.find_parameters()
  1083. for param in find_parameters(self, ['job_properties']):
  1084. params[param] = ''
  1085. for dataset in self.dataset_set.all():
  1086. for param in find_parameters(dataset, ['uri']):
  1087. if param not in set(DATASET_FREQUENCY):
  1088. params[param] = ''
  1089. for ds in self.datainput_set.all():
  1090. params.pop(ds.name, None)
  1091. for ds in self.dataoutput_set.all():
  1092. params.pop(ds.name, None)
  1093. for wf_param in json.loads(self.job_properties):
  1094. params.pop(wf_param['name'], None)
  1095. return params
  1096. def utc_datetime_format(utc_time):
  1097. return utc_time.strftime("%Y-%m-%dT%H:%MZ")
  1098. class DatasetManager(models.Manager):
  1099. def can_read_or_exception(self, request, dataset_id):
  1100. if dataset_id is None:
  1101. return
  1102. try:
  1103. dataset = Dataset.objects.get(pk=dataset_id)
  1104. if dataset.coordinator.can_read(request.user):
  1105. return dataset
  1106. else:
  1107. message = _("Permission denied. %(username)s does not have the permissions to access dataset %(id)s.") % \
  1108. {'username': request.user.username, 'id': dataset.id}
  1109. access_warn(request, message)
  1110. request.error(message)
  1111. raise PopupException(message)
  1112. except Dataset.DoesNotExist:
  1113. raise PopupException(_('dataset %(id)s not exist') % {'id': dataset_id})
  1114. class Dataset(models.Model):
  1115. """
  1116. http://oozie.apache.org/docs/3.3.0/CoordinatorFunctionalSpec.html#a6.3._Synchronous_Coordinator_Application_Definition
  1117. """
  1118. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name'),
  1119. help_text=_t('The name of the dataset.'))
  1120. description = models.CharField(max_length=1024, blank=True, default='', verbose_name=_t('Description'),
  1121. help_text=_t('A description of the dataset.'))
  1122. start = models.DateTimeField(default=datetime.today(), verbose_name=_t('Start'),
  1123. help_text=_t(' The UTC datetime of the initial instance of the dataset. The initial instance also provides '
  1124. 'the baseline datetime to compute instances of the dataset using multiples of the frequency.'))
  1125. frequency_number = models.SmallIntegerField(default=1, choices=FREQUENCY_NUMBERS, verbose_name=_t('Frequency number'),
  1126. help_text=_t('The number of units of the rate at which '
  1127. 'data is periodically created.'))
  1128. frequency_unit = models.CharField(max_length=20, choices=FREQUENCY_UNITS, default='days', verbose_name=_t('Frequency unit'),
  1129. help_text=_t('The unit of the rate at which data is periodically created.'))
  1130. uri = models.CharField(max_length=1024, default='/data/${YEAR}${MONTH}${DAY}', verbose_name=_t('URI'),
  1131. help_text=_t('The URI template that identifies the dataset and can be resolved into concrete URIs to identify a particular '
  1132. 'dataset instance. The URI consist of constants (e.g. ${YEAR}/${MONTH}) and '
  1133. 'configuration properties (e.g. /home/${USER}/projects/${PROJECT})'))
  1134. timezone = models.CharField(max_length=24, choices=TIMEZONES, default='America/Los_Angeles', verbose_name=_t('Timezone'),
  1135. help_text=_t('The timezone of the dataset.'))
  1136. done_flag = models.CharField(max_length=64, blank=True, default='', verbose_name=_t('Done flag'),
  1137. help_text=_t('The done file for the data set. If the Done flag is not specified, then Oozie '
  1138. 'configures Hadoop to create a _SUCCESS file in the output directory. If Done '
  1139. 'flag is set to empty, then Coordinator looks for the existence of the directory itself.'))
  1140. coordinator = models.ForeignKey(Coordinator, verbose_name=_t('Coordinator'),
  1141. help_text=_t('The coordinator associated with this data.'))
  1142. instance_choice = models.CharField(max_length=10, default='default', verbose_name=_t('Instance type'),
  1143. help_text=_t('Customize the date instance(s), e.g. define a range of dates, use EL functions...'))
  1144. advanced_start_instance = models.CharField(max_length=128, default='0', verbose_name=_t('Start instance'),
  1145. help_text=_t('Shift the frequency for gettting past/future start date or enter verbatim the Oozie start instance, e.g. ${coord:current(0)}'))
  1146. advanced_end_instance = models.CharField(max_length=128, blank=True, default='0', verbose_name=_t('End instance'),
  1147. help_text=_t('Optional: Shift the frequency for gettting past/future end dates or enter verbatim the Oozie end instance.'))
  1148. objects = DatasetManager()
  1149. unique_together = ('coordinator', 'name')
  1150. def __unicode__(self):
  1151. return '%s' % (self.name,)
  1152. @property
  1153. def start_utc(self):
  1154. return utc_datetime_format(self.start)
  1155. @property
  1156. def frequency(self):
  1157. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1158. @property
  1159. def text_frequency(self):
  1160. return '%(number)d %(unit)s' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1161. @property
  1162. def start_instance(self):
  1163. if not self.is_advanced_start_instance:
  1164. return int(self.advanced_start_instance)
  1165. else:
  1166. return 0
  1167. @property
  1168. def is_advanced_start_instance(self):
  1169. return not self.is_int(self.advanced_start_instance)
  1170. def is_int(self, text):
  1171. try:
  1172. int(text)
  1173. return True
  1174. except ValueError:
  1175. return False
  1176. @property
  1177. def end_instance(self):
  1178. if not self.is_advanced_end_instance:
  1179. return int(self.advanced_end_instance)
  1180. else:
  1181. return 0
  1182. @property
  1183. def is_advanced_end_instance(self):
  1184. return not self.is_int(self.advanced_end_instance)
  1185. class DataInput(models.Model):
  1186. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name of an input variable in the workflow.'),
  1187. help_text=_t('The name of the variable of the workflow to automatically fill up.'))
  1188. dataset = models.OneToOneField(Dataset, verbose_name=_t('The dataset representing format of the data input.'),
  1189. help_text=_t('The pattern of the input data we want to process.'))
  1190. coordinator = models.ForeignKey(Coordinator)
  1191. unique_together = ('coordinator', 'name')
  1192. class DataOutput(models.Model):
  1193. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name of an output variable in the workflow'),
  1194. help_text=_t('The name of the variable of the workflow to automatically filled up.'))
  1195. dataset = models.OneToOneField(Dataset, verbose_name=_t('The dataset representing the format of the data output.'),
  1196. help_text=_t('The pattern of the output data we want to generate.'))
  1197. coordinator = models.ForeignKey(Coordinator)
  1198. unique_together = ('coordinator', 'name')
  1199. class BundledCoordinator(models.Model):
  1200. bundle = models.ForeignKey('Bundle', verbose_name=_t('Bundle'),
  1201. help_text=_t('The bundle regrouping all the coordinators.'))
  1202. coordinator = models.ForeignKey(Coordinator, verbose_name=_t('Coordinator'),
  1203. help_text=_t('The coordinator to batch with other coordinators.'))
  1204. parameters = models.TextField(default='[{"name":"oozie.use.system.libpath","value":"true"}]', verbose_name=_t('Parameters'),
  1205. help_text=_t('Constants used at the submission time (e.g. market=US, oozie.use.system.libpath=true).'))
  1206. def get_parameters(self):
  1207. return json.loads(self.parameters)
  1208. class Bundle(Job):
  1209. kick_off_time = models.DateTimeField(default=datetime.today(), verbose_name=_t('Start'),
  1210. help_text=_t('When to start the first coordinators.'))
  1211. coordinators = models.ManyToManyField(Coordinator, through='BundledCoordinator')
  1212. HUE_ID = 'hue-id-b'
  1213. ICON = '/oozie/static/art/icon_oozie_24_bundle.png'
  1214. def get_type(self):
  1215. return 'bundle'
  1216. def to_xml(self, mapping=None):
  1217. if mapping is None:
  1218. mapping = {}
  1219. tmpl = "editor/gen/bundle.xml.mako"
  1220. return force_unicode(
  1221. re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
  1222. 'bundle': self,
  1223. 'mapping': mapping
  1224. })))
  1225. def clone(self, new_owner=None):
  1226. bundleds = BundledCoordinator.objects.filter(bundle=self)
  1227. copy_doc = self.doc.get()
  1228. copy = self
  1229. copy.pk = None
  1230. copy.id = None
  1231. copy.name += '-copy'
  1232. copy.deployment_dir = ''
  1233. if new_owner is not None:
  1234. copy.owner = new_owner
  1235. copy.save()
  1236. for bundled in bundleds:
  1237. bundled.pk = None
  1238. bundled.id = None
  1239. bundled.bundle = copy
  1240. bundled.save()
  1241. copy_doc.pk = None
  1242. copy_doc.id = None
  1243. copy_doc.owner = copy.owner
  1244. copy_doc.save()
  1245. copy.doc.add(copy_doc)
  1246. return copy
  1247. @classmethod
  1248. def get_application_path_key(cls):
  1249. return 'oozie.bundle.application.path'
  1250. @classmethod
  1251. def get_application_filename(cls):
  1252. return 'bundle.xml'
  1253. def get_absolute_url(self):
  1254. return reverse('oozie:edit_bundle', kwargs={'bundle': self.id})
  1255. def find_parameters(self):
  1256. params = {}
  1257. for bundled in BundledCoordinator.objects.filter(bundle=self):
  1258. for param in bundled.coordinator.find_parameters():
  1259. params[param] = ''
  1260. for param in bundled.get_parameters():
  1261. params.pop(param['name'], None)
  1262. return params
  1263. @property
  1264. def kick_off_time_utc(self):
  1265. return utc_datetime_format(self.kick_off_time)
  1266. class HistoryManager(models.Manager):
  1267. def create_from_submission(self, submission):
  1268. History.objects.create(submitter=submission.user,
  1269. oozie_job_id=submission.oozie_id,
  1270. job=submission.job,
  1271. properties=json.dumps(submission.properties))
  1272. class History(models.Model):
  1273. """
  1274. Contains information on submitted workflows/coordinators.
  1275. """
  1276. submitter = models.ForeignKey(User, db_index=True)
  1277. submission_date = models.DateTimeField(auto_now=True, db_index=True)
  1278. oozie_job_id = models.CharField(max_length=128)
  1279. job = models.ForeignKey(Job, db_index=True)
  1280. properties = models.TextField()
  1281. objects = HistoryManager()
  1282. @property
  1283. def properties_dict(self):
  1284. return json.loads(self.properties)
  1285. def get_absolute_oozie_url(self):
  1286. view = 'oozie:list_oozie_workflow'
  1287. if self.oozie_job_id.endswith('C'):
  1288. view = 'oozie:list_oozie_coordinator'
  1289. elif self.oozie_job_id.endswith('B'):
  1290. view = 'oozie:list_oozie_bundle'
  1291. return reverse(view, kwargs={'job_id': self.oozie_job_id})
  1292. def get_workflow(self):
  1293. if self.oozie_job_id.endswith('W'):
  1294. return self.job.get_full_node()
  1295. def get_coordinator(self):
  1296. if self.oozie_job_id.endswith('C'):
  1297. return self.job.get_full_node()
  1298. @classmethod
  1299. def get_workflow_from_config(self, conf_dict):
  1300. try:
  1301. return Workflow.objects.get(id=conf_dict.get(Workflow.HUE_ID))
  1302. except Workflow.DoesNotExist:
  1303. pass
  1304. @classmethod
  1305. def get_coordinator_from_config(self, conf_dict):
  1306. try:
  1307. return Coordinator.objects.get(id=conf_dict.get(Coordinator.HUE_ID))
  1308. except Coordinator.DoesNotExist:
  1309. pass
  1310. @classmethod
  1311. def cross_reference_submission_history(cls, user, oozie_id, coordinator_job_id):
  1312. # Try do get the history
  1313. history = None
  1314. try:
  1315. history = History.objects.get(oozie_job_id=oozie_id)
  1316. if history.job.owner != user:
  1317. history = None
  1318. except History.DoesNotExist:
  1319. pass
  1320. return history
  1321. def find_parameters(instance, fields=None):
  1322. """Find parameters in the given fields"""
  1323. if fields is None:
  1324. fields = [field.name for field in instance._meta.fields]
  1325. params = []
  1326. for field in fields:
  1327. data = getattr(instance, field)
  1328. if isinstance(data, basestring):
  1329. for match in Template.pattern.finditer(data):
  1330. name = match.group('braced')
  1331. if name is not None:
  1332. params.append(name)
  1333. return params
  1334. # See http://wiki.apache.org/hadoop/JobConfFile
  1335. _STD_PROPERTIES = [
  1336. 'mapred.input.dir',
  1337. 'mapred.output.dir',
  1338. 'mapred.job.name',
  1339. 'mapred.job.queue.name',
  1340. 'mapred.mapper.class',
  1341. 'mapred.reducer.class',
  1342. 'mapred.combiner.class',
  1343. 'mapred.partitioner.class',
  1344. 'mapred.map.tasks',
  1345. 'mapred.reduce.tasks',
  1346. 'mapred.input.format.class',
  1347. 'mapred.output.format.class',
  1348. 'mapred.input.key.class',
  1349. 'mapred.input.value.class',
  1350. 'mapred.output.key.class',
  1351. 'mapred.output.value.class',
  1352. 'mapred.mapoutput.key.class',
  1353. 'mapred.mapoutput.value.class',
  1354. 'mapred.combine.buffer.size',
  1355. 'mapred.min.split.size',
  1356. 'mapred.speculative.execution',
  1357. 'mapred.map.tasks.speculative.execution',
  1358. 'mapred.reduce.tasks.speculative.execution',
  1359. 'mapred.queue.default.acl-administer-jobs',
  1360. ]
  1361. _STD_PROPERTIES_JSON = json.dumps(_STD_PROPERTIES)
  1362. ACTION_TYPES = {
  1363. Mapreduce.node_type: Mapreduce,
  1364. Streaming.node_type: Streaming,
  1365. Java.node_type: Java,
  1366. Pig.node_type: Pig,
  1367. Hive.node_type: Hive,
  1368. Sqoop.node_type: Sqoop,
  1369. Ssh.node_type: Ssh,
  1370. Shell.node_type: Shell,
  1371. DistCp.node_type: DistCp,
  1372. Fs.node_type: Fs,
  1373. Email.node_type: Email,
  1374. SubWorkflow.node_type: SubWorkflow,
  1375. Generic.node_type: Generic,
  1376. }
  1377. CONTROL_TYPES = {
  1378. Fork.node_type: Fork,
  1379. Join.node_type: Join,
  1380. Decision.node_type: Decision,
  1381. DecisionEnd.node_type: DecisionEnd,
  1382. Start.node_type: Start,
  1383. End.node_type: End,
  1384. }
  1385. NODE_TYPES = ACTION_TYPES.copy()
  1386. NODE_TYPES.update(CONTROL_TYPES)