Browse Source

[oozie] Add prepare statement in a workflow action

Add new field 'prepares' to Pig, MapReduce, Java actions
Update XML generation
Add tests
Fix 'Add Variable' js bug
Romain Rigaux 13 năm trước cách đây
mục cha
commit
219d5bd

+ 3 - 0
apps/oozie/src/oozie/forms.py

@@ -61,6 +61,7 @@ class JavaForm(forms.ModelForm):
     exclude = NodeForm.Meta.ALWAYS_HIDE
     widgets = {
       'job_properties': forms.widgets.HiddenInput(),
+      'prepares': forms.widgets.HiddenInput(),
       'files': forms.HiddenInput(),
       'archives': forms.HiddenInput(),
       'jar_path': forms.TextInput(attrs={'class': 'pathChooser span5'}),
@@ -75,6 +76,7 @@ class MapreduceForm(forms.ModelForm):
     exclude = NodeForm.Meta.ALWAYS_HIDE
     widgets = {
       'job_properties': forms.widgets.HiddenInput(),
+      'prepares': forms.widgets.HiddenInput(),
       'files': forms.HiddenInput(),
       'archives': forms.HiddenInput(),
       'jar_path': forms.TextInput(attrs={'class': 'pathChooser span5'}),
@@ -101,6 +103,7 @@ class PigForm(forms.ModelForm):
     exclude = NodeForm.Meta.ALWAYS_HIDE
     widgets = {
       'job_properties': forms.widgets.HiddenInput(),
+      'prepares': forms.widgets.HiddenInput(),
       'params': forms.widgets.HiddenInput(),
       'script_path': forms.TextInput(attrs={'class': 'pathChooser span5'}),
       'files': forms.widgets.HiddenInput(),

+ 16 - 4
apps/oozie/src/oozie/models.py

@@ -666,7 +666,7 @@ class Action(Node):
 #  - Node.get_full_node()
 
 class Mapreduce(Action):
-  PARAM_FIELDS = ('files', 'archives', 'job_properties', 'jar_path')
+  PARAM_FIELDS = ('files', 'archives', 'job_properties', 'jar_path', 'prepares')
   node_type = 'mapreduce'
 
   files = models.CharField(max_length=PATH_MAX, default="[]",
@@ -676,6 +676,7 @@ class Mapreduce(Action):
   job_properties = models.TextField(default='[]', # JSON dict
                                     help_text=_t('For the job configuration (e.g. mapred.mapper.class)'))
   jar_path = models.CharField(max_length=PATH_MAX, help_text=_t('Path to jar files on HDFS'))
+  prepares = models.TextField(default="[]", help_text=_t('List of paths to delete of create before starting the job'))
 
   def get_properties(self):
     return json.loads(self.job_properties)
@@ -686,6 +687,9 @@ class Mapreduce(Action):
   def get_archives(self):
     return json.loads(self.archives)
 
+  def get_prepares(self):
+    return json.loads(self.prepares)
+
 
 class Streaming(Action):
   PARAM_FIELDS = ('files', 'archives', 'job_properties', 'mapper', 'reducer')
@@ -710,7 +714,7 @@ class Streaming(Action):
 
 class Java(Action):
   PARAM_FIELDS = ('files', 'archives', 'jar_path', 'main_class', 'args',
-                  'java_opts', 'job_properties')
+                  'java_opts', 'job_properties', 'prepares')
   node_type = "java"
 
   files = models.CharField(max_length=PATH_MAX, default="[]",
@@ -723,6 +727,7 @@ class Java(Action):
   java_opts = models.CharField(max_length=256, blank=True)
   job_properties = models.TextField(default='[]', # JSON dict
                                     help_text=_t('For the job configuration (e.g. mapred.mapper.class)'))
+  prepares = models.TextField(default="[]", help_text=_t('List of paths to delete of create before starting the job'))
 
   def get_properties(self):
     return json.loads(self.job_properties)
@@ -733,13 +738,16 @@ class Java(Action):
   def get_archives(self):
     return json.loads(self.archives)
 
+  def get_prepares(self):
+    return json.loads(self.prepares)
+
 
 class Pig(Action):
-  PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params')
+  PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
   node_type = 'pig'
 
   script_path = models.CharField(max_length=256, blank=False, help_text=_t('Local path'))
-  params = models.TextField(default="[]")
+  params = models.TextField(default="[]", help_text=_t('The Pig parameters of the script'))
 
   files = models.CharField(max_length=PATH_MAX, default="[]",
       help_text=_t('List of paths to files to be added to the distributed cache'))
@@ -747,6 +755,7 @@ class Pig(Action):
       help_text=_t('List of paths to archives to be added to the distributed cache'))
   job_properties = models.TextField(default='[{"name":"oozie.use.system.libpath","value":"true"}]', # JSON dict
                                     help_text=_t('For the job configuration (e.g. mapred.mapper.class)'))
+  prepares = models.TextField(default="[]", help_text=_t('List of paths to delete of create before starting the job'))
 
   def get_properties(self):
     return json.loads(self.job_properties)
@@ -760,6 +769,9 @@ class Pig(Action):
   def get_params(self):
     return json.loads(self.params)
 
+  def get_prepares(self):
+    return json.loads(self.prepares)
+
 
 Action.types = (Mapreduce.node_type, Streaming.node_type, Java.node_type, Pig.node_type)
 

+ 6 - 6
apps/oozie/src/oozie/templates/editor/edit_coordinator.mako

@@ -253,11 +253,11 @@ ${ layout.menubar(section='coordinators') }
                     </tr>
 
                      <div class="hide">
-                        % for field in form.visible_fields():
-                            ${ field.errors }
-                            ${ field.label }: ${ field }
-                        % endfor
-                        </div>
+                       % for field in form.visible_fields():
+                          ${ field.errors }
+                          ${ field.label }: ${ field }
+                       % endfor
+                     </div>
 
                    % endfor
                   </tbody>
@@ -387,7 +387,7 @@ ${ layout.menubar(section='coordinators') }
 
    $("a[data-row-selector='true']").jHueRowSelector();
 
-   ko.applyBindings(window.viewModel)
+   ko.applyBindings(window.viewModel);
  });
 </script>
 

+ 6 - 4
apps/oozie/src/oozie/templates/editor/edit_workflow.mako

@@ -180,10 +180,10 @@ ${ layout.menubar(section='workflows') }
 
 
 <style type="text/css">
-#modal-window .modal-content {
-  height: 300px;
-  overflow: auto;
-}
+  #modal-window .modal-content {
+    height: 300px;
+    overflow: auto;
+  }
 </style>
 
 <script src="/static/ext/js/knockout-2.0.0.js" type="text/javascript" charset="utf-8"></script>
@@ -210,6 +210,8 @@ ${ layout.menubar(section='workflows') }
       });
     });
 
+    ko.applyBindings(window.viewModel);
+
     $("a[data-row-selector='true']").jHueRowSelector();
   });
 </script>

+ 58 - 3
apps/oozie/src/oozie/templates/editor/edit_workflow_action.mako

@@ -61,6 +61,43 @@ ${ layout.menubar(section='workflows') }
       % endif
     % endfor
 
+    % if 'prepares' in action_form.fields:
+      <div class="control-group">
+        <label class="control-label">${ _('Prepare') }</label>
+        <div class="controls">
+          <table class="table-condensed designTable" data-bind="visible: prepares().length > 0">
+            <thead>
+              <tr>
+                <th>${ _('Type') }</th>
+                <th>${ _('Value') }</th>
+                <th/>
+              </tr>
+            </thead>
+            <tbody data-bind="foreach: prepares">
+              <tr>
+                <td>
+                  <span class="span3 required" data-bind="text: type" />
+                </td>
+                <td>
+                  <input class="input span5 required pathChooserKo" data-bind="fileChooser: $data, value: value, uniqueName: false" />
+                </td>
+                <td><a class="btn" href="#" data-bind="click: $root.removePrepare">${ _('Delete') }</a></td>
+              </tr>
+            </tbody>
+          </table>
+
+          % if len(action_form['prepares'].errors):
+            <div class="alert alert-error">
+              ${ unicode(action_form['prepares'].errors) | n }
+            </div>
+          % endif
+
+          <button class="btn" data-bind="click: addPrepareDelete">${ _('Add delete') }</button>
+          <button class="btn" data-bind="click: addPrepareMkdir">${ _('Add mkdir') }</button>
+        </div>
+      </div>
+    % endif
+
     % if 'params' in action_form.fields:
       <div class="control-group">
         <label class="control-label">${ _('Params') }</label>
@@ -275,13 +312,14 @@ ${ layout.menubar(section='workflows') }
         });
     };
 
-    var ViewModel = function(properties, files, archives, params) {
+    var ViewModel = function(properties, files, archives, params, prepares) {
         var self = this;
 
         self.properties = ko.observableArray(properties);
         self.files = ko.observableArray(files);
         self.archives = ko.observableArray(archives);
         self.params = ko.observableArray(params);
+        self.prepares = ko.observableArray(prepares);
 
         self.addProp = function() {
             self.properties.push({ name: "", value: "" });
@@ -304,6 +342,18 @@ ${ layout.menubar(section='workflows') }
             self.params.remove(val);
         };
 
+        self.addPrepareDelete = function() {
+            self.prepares.push({ value: "", type: "delete" });
+        };
+
+        self.addPrepareMkdir = function() {
+            self.prepares.push({ value: "", type: "mkdir" });
+        };
+
+        self.removePrepare = function(val) {
+            self.prepares.remove(val);
+        };
+
         self.addFile = function() {
             self.files.push({ name: "", dummy: "" });
         };
@@ -325,7 +375,7 @@ ${ layout.menubar(section='workflows') }
             var files_arr = dictArrayToArray(ko.toJS(self.files));
             var archives_arr = dictArrayToArray(ko.toJS(self.archives));
 
-            // Beware dirty
+            // Beware: dirty
             $("<input>").attr("type", "hidden")
                 .attr("name", "job_properties")
                 .attr("value", ko.utils.stringifyJson(self.properties))
@@ -342,6 +392,10 @@ ${ layout.menubar(section='workflows') }
                 .attr("name", "params")
                 .attr("value", ko.utils.stringifyJson(self.params))
                 .appendTo(form);
+            $("<input>").attr("type", "hidden")
+                .attr("name", "prepares")
+                .attr("value", ko.utils.stringifyJson(self.prepares))
+                .appendTo(form);
 
             form.submit();
         };
@@ -351,7 +405,8 @@ ${ layout.menubar(section='workflows') }
               ${ job_properties },
               arrayToDictArray(${ files }),
               arrayToDictArray(${ archives }),
-              ${ params });
+              ${ params },
+              ${ prepares });
 
     ko.bindingHandlers.fileChooser = {
           init: function(element, valueAccessor, allBindings, model) {

+ 11 - 0
apps/oozie/src/oozie/templates/editor/gen/workflow-common.xml.mako

@@ -26,6 +26,17 @@ import posixpath
 <%def name="filelink(path)">${ path + '#' + posixpath.basename(path) }</%def>
 
 
+<%def name="prepares(prepares)">
+        % if prepares:
+            <prepare>
+                % for p in prepares:
+                <${ p['type'] } path="${ p['value'] }"/>
+                % endfor
+            </prepare>
+        % endif
+</%def>
+
+
 <%def name="configuration(properties)">
         % if properties:
             <configuration>

+ 1 - 0
apps/oozie/src/oozie/templates/editor/gen/workflow-java.xml.mako

@@ -21,6 +21,7 @@
             <job-tracker>${'${'}jobTracker}</job-tracker>
             <name-node>${'${'}nameNode}</name-node>
 
+            ${ common.prepares(node.get_prepares()) }
             ${ common.configuration(node.get_properties()) }
 
             <main-class>${ node.main_class }</main-class>

+ 1 - 0
apps/oozie/src/oozie/templates/editor/gen/workflow-mapreduce.xml.mako

@@ -21,6 +21,7 @@
             <job-tracker>${'${'}jobTracker}</job-tracker>
             <name-node>${'${'}nameNode}</name-node>
 
+            ${ common.prepares(node.get_prepares()) }
             ${ common.configuration(node.get_properties()) }
 
             ${ common.distributed_cache(node.get_files(), node.get_archives()) }

+ 1 - 0
apps/oozie/src/oozie/templates/editor/gen/workflow-pig.xml.mako

@@ -21,6 +21,7 @@
             <job-tracker>${'${'}jobTracker}</job-tracker>
             <name-node>${'${'}nameNode}</name-node>
 
+            ${ common.prepares(node.get_prepares()) }
             ${ common.configuration(node.get_properties()) }
 
             <script>${ node.script_path }</script>

+ 5 - 0
apps/oozie/src/oozie/templates/editor/import_workflow_action.mako

@@ -45,6 +45,11 @@
               <td>${ action.description }</td>
             </tr>
           %endfor
+          % if not available_actions:
+            <tr class="action-row">
+              <td>${ _('N/A') }</td><td></td><td></td>
+            </tr>
+          % endif
         </tbody>
       </table>
     </div>

+ 1 - 1
apps/oozie/src/oozie/templates/editor/job_action_properties.mako

@@ -15,7 +15,7 @@
 ## limitations under the License.
 
 <%!
-from django.utils.translation import ugettext as _
+  from django.utils.translation import ugettext as _
 %>
 
 <%namespace name="utils" file="../../utils.inc.mako" />

+ 14 - 4
apps/oozie/src/oozie/tests.py

@@ -131,7 +131,8 @@ class TestEditor:
 
 
   def test_find_all_parameters(self):
-        assert_equal([{'name': u'SLEEP', 'value': ''}, {'name': u'market', 'value': u'US'}], self.wf.find_all_parameters())
+        assert_equal([{'name': u'output', 'value': u''}, {'name': u'SLEEP', 'value': ''}, {'name': u'market', 'value': u'US'}],
+                     self.wf.find_all_parameters())
 
 
   def test_move_up(self):
@@ -342,6 +343,9 @@ class TestEditor:
         '        <map-reduce>\n'
         '           <job-tracker>${jobTracker}</job-tracker>\n'
         '            <name-node>${nameNode}</name-node>\n'
+        '            <prepare>\n'
+        '                <delete path="${output}"/>\n'
+        '            </prepare>\n'
         '            <configuration>\n'
         '                <property>\n'
         '                    <name>sleep</name>\n'
@@ -356,6 +360,9 @@ class TestEditor:
         '        <map-reduce>\n'
         '            <job-tracker>${jobTracker}</job-tracker>\n'
         '            <name-node>${nameNode}</name-node>\n'
+        '            <prepare>\n'
+        '                <delete path="${output}"/>\n'
+        '            </prepare>\n'
         '            <configuration>\n'
         '                <property>\n'
         '                    <name>sleep</name>\n'
@@ -370,6 +377,9 @@ class TestEditor:
         '        <map-reduce>\n'
         '            <job-tracker>${jobTracker}</job-tracker>\n'
         '            <name-node>${nameNode}</name-node>\n'
+        '            <prepare>\n'
+        '                <delete path="${output}"/>\n'
+        '            </prepare>\n'
         '            <configuration>\n'
         '                <property>\n'
         '                    <name>sleep</name>\n'
@@ -814,7 +824,8 @@ class TestEditor:
 
 
 # Utils
-WORKFLOW_DICT = {u'deployment_dir': [u''], u'name': [u'wf-name-1'], u'description': [u''], u'parameters': [u'[{"name":"market","value":"US"}]']}
+WORKFLOW_DICT = {u'deployment_dir': [u''], u'name': [u'wf-name-1'], u'description': [u''],
+                 u'parameters': [u'[{"name":"market","value":"US"}]']}
 
 
 # Beware: client not consistent with self.c in TestEditor
@@ -823,8 +834,7 @@ def add_action(workflow, action, name):
 
   response = c.post("/oozie/new_action/%s/%s/%s" % (workflow, 'mapreduce', action), {
      u'files': [u'[]'], u'name': [name], u'jar_path': [u'/tmp/.file.jar'], u'job_properties': [u'[{"name":"sleep","value":"${SLEEP}"}]'],
-     u'archives': [u'[]'], u'description': [u'']}, follow=True)
-  assert_equal(200, response.status_code)
+     u'archives': [u'[]'], u'description': [u''], u'prepares': [u'[{"type":"delete","value":"${output}"}]']}, follow=True)
   assert_true(Node.objects.filter(name=name).exists(), response)
   return Node.objects.get(name=name)
 

+ 2 - 0
apps/oozie/src/oozie/views/editor.py

@@ -432,6 +432,7 @@ def new_action(request, workflow, node_type, parent_action_id):
       'files': extract_field_data(action_form['files']),
       'archives': extract_field_data(action_form['archives']),
       'params': 'params' in action_form.fields and extract_field_data(action_form['params']) or '[]',
+      'prepares': 'prepares' in action_form.fields and extract_field_data(action_form['prepares']) or '[]',
       'action_form': action_form,
       'node_type': node_type,
       'properties_hint': _STD_PROPERTIES_JSON,
@@ -460,6 +461,7 @@ def edit_action(request, action):
     'files': extract_field_data(action_form['files']),
     'archives': extract_field_data(action_form['archives']),
     'params': 'params' in action_form.fields and extract_field_data(action_form['params']) or '[]',
+    'prepares': 'prepares' in action_form.fields and extract_field_data(action_form['prepares']) or '[]',
     'action_form': action_form,
     'node_type': action.node_type,
     'properties_hint': _STD_PROPERTIES_JSON,