|
|
@@ -81,12 +81,13 @@ class Submission(object):
|
|
|
wf_xml = self._generate_workflow_xml(fs_defaultfs)
|
|
|
self._do_as(self._username, self._copy_files, wf_dir, wf_xml)
|
|
|
LOG.info("Prepared deployment directory at '%s' for %s" % (wf_dir, self))
|
|
|
+ LOG.info("Submitting design id %s to %s as `%s'" % (self._design_obj.id, jobtracker, self._username))
|
|
|
|
|
|
try:
|
|
|
prev = get_oozie().setuser(self._username)
|
|
|
self._job_id = get_oozie().submit_workflow(
|
|
|
self._fs.get_hdfs_path(wf_dir),
|
|
|
- properties={ 'jobTracker': jobtracker })
|
|
|
+ properties=self._get_properties(jobtracker))
|
|
|
LOG.info("Submitted: %s" % (self,))
|
|
|
|
|
|
# Now we need to run it
|
|
|
@@ -96,7 +97,15 @@ class Submission(object):
|
|
|
get_oozie().setuser(prev)
|
|
|
|
|
|
return self.job_id
|
|
|
-
|
|
|
+
|
|
|
+
|
|
|
+ def _get_properties(self, jobtracker_addr):
|
|
|
+ res = { 'jobTracker': jobtracker_addr }
|
|
|
+ if self._design_obj.get_root_action().action_type == \
|
|
|
+ models.OozieStreamingAction.ACTION_TYPE:
|
|
|
+ res['oozie.use.system.libpath'] = 'true'
|
|
|
+ return res
|
|
|
+
|
|
|
|
|
|
def _copy_files(self, wf_dir, wf_xml):
|
|
|
"""
|