瀏覽代碼

HUE-633. [jobsub] Port examples to new design models

Adding:
- Streaming wordcount
- MR sleep job
- MR grep job
bc Wong 13 年之前
父節點
當前提交
fda9f479a4

二進制
apps/jobsub/data/examples/hadoop-examples.jar


+ 8 - 2
apps/jobsub/src/jobsub/conf.py

@@ -19,6 +19,7 @@
 import os.path
 
 from desktop.lib.conf import Config, coerce_bool
+from desktop.lib import paths
 
 REMOTE_DATA_DIR = Config(
   key="remote_data_dir",
@@ -30,8 +31,13 @@ LOCAL_DATA_DIR = Config(
   key="local_data_dir",
   default=os.path.join(os.path.dirname(__file__), "..", "..", "data"),
   help="Location on local FS where examples and template are stored",
-  private=True
-)
+  private=True)
+
+SAMPLE_DATA_DIR = Config(
+  key="sample_data_dir",
+  default=paths.get_thirdparty_root("sample_data"),
+  help="Location on local FS where sample data is stored",
+  private=True)
 
 OOZIE_URL = Config(
   key='oozie_url',

+ 43 - 18
apps/jobsub/src/jobsub/fixtures/example_data.xml

@@ -15,28 +15,53 @@
     <field to="auth.group" name="groups" rel="ManyToManyRel"></field>
     <field to="auth.permission" name="user_permissions" rel="ManyToManyRel"></field>
   </object>
-  <object pk="1" model="jobsub.jobdesign">
+  <object pk="1" model="jobsub.oozieaction">
+    <field type="CharField" name="action_type">streaming</field>
+  </object>
+  <object pk="2" model="jobsub.oozieaction">
+    <field type="CharField" name="action_type">mapreduce</field>
+  </object>
+  <object pk="3" model="jobsub.oozieaction">
+    <field type="CharField" name="action_type">mapreduce</field>
+  </object>
+  <object pk="1" model="jobsub.ooziedesign">
     <field to="auth.user" name="owner" rel="ManyToOneRel">1</field>
-    <field type="CharField" name="name">Example: Pi Calculator</field>
-    <field type="CharField" name="description">Hadoop's built-in Pi example.  This job commonly fails because /user/&lt;your-user-name&gt; is not writable.  iterations_per_mapper is how many random samples to take per-mapper.  number_of_mappers is how many mappers to launch.</field>
-    <field type="DateField" name="last_modified">2009-09-21</field>
-    <field type="CharField" name="type">jar</field>
-    <field type="CharField" name="data">{"arguments": "pi $num_of_mappers $iterations_per_mapper", "jarfile": "/user/hue/jobsub/examples/hadoop-examples.jar"}</field>
+    <field type="CharField" name="name">streaming_wordcount</field>
+    <field type="CharField" name="description">[Sample] Wordcount</field>
+    <field type="DateTimeField" name="last_modified">2012-04-02 01:13:31</field>
+    <field to="jobsub.oozieaction" name="root_action" rel="ManyToOneRel">1</field>
   </object>
-  <object pk="2" model="jobsub.jobdesign">
+  <object pk="2" model="jobsub.ooziedesign">
     <field to="auth.user" name="owner" rel="ManyToOneRel">1</field>
-    <field type="CharField" name="name">Example: Streaming Wordcount</field>
-    <field type="CharField" name="description">Runs wordcount, implemented in python, via streaming.</field>
-    <field type="DateField" name="last_modified">2009-09-21</field>
-    <field type="CharField" name="type">streaming</field>
-    <field type="CharField" name="data">{"mapper_cmd": "python wordcount.py map", "inputformat_class": "org.apache.hadoop.mapred.TextInputFormat", "cache_archives": [], "reducer_class": "", "num_reduce_tasks": 1, "cache_files": ["/user/hue/jobsub/examples/wordcount.py"], "inputreader": "", "combiner_class": "", "outputformat_class": "org.apache.hadoop.mapred.TextOutputFormat", "mapper_class": "", "output": "$output", "input": ["/user/hue/jobsub/sample_data/sonnets.txt", "/user/hue/jobsub/sample_data/midsummer.txt"], "hadoop_properties": {}, "reducer_cmd": "python wordcount.py reduce", "partitioner_class": ""}</field>
+    <field type="CharField" name="name">grep_example</field>
+    <field type="CharField" name="description">[Sample] Grep for "dream" in some Shakespearen text</field>
+    <field type="DateTimeField" name="last_modified">2012-04-02 02:04:34</field>
+    <field to="jobsub.oozieaction" name="root_action" rel="ManyToOneRel">2</field>
   </object>
-  <object pk="3" model="jobsub.jobdesign">
+  <object pk="3" model="jobsub.ooziedesign">
     <field to="auth.user" name="owner" rel="ManyToOneRel">1</field>
-    <field type="CharField" name="name">Example: Sleep Job</field>
-    <field type="CharField" name="description">MapReduce's sleep example</field>
-    <field type="DateTimeField" name="last_modified">2009-09-26 20:32:03</field>
-    <field type="CharField" name="type">jar</field>
-    <field type="CharField" name="data">{"arguments": "sleep -m $num_mappers -r $num_reducers -mt $map_sleep_time_millis -rt $reduce_sleep_time_millis", "jarfile": "/user/hue/jobsub/examples/hadoop-examples.jar"}</field>
+    <field type="CharField" name="name">sleep_job</field>
+    <field type="CharField" name="description">[Sample] Sleep Job</field>
+    <field type="DateTimeField" name="last_modified">2012-04-02 03:06:02</field>
+    <field to="jobsub.oozieaction" name="root_action" rel="ManyToOneRel">3</field>
+  </object>
+  <object pk="2" model="jobsub.ooziemapreduceaction">
+    <field type="CharField" name="files">[]</field>
+    <field type="CharField" name="archives">[]</field>
+    <field type="CharField" name="job_properties">[{"name":"mapred.mapper.regex","value":"dream"},{"name":"mapred.input.dir","value":"/user/hue/jobsub/sample_data"},{"name":"mapred.output.dir","value":"$output_dir"},{"name":"mapred.mapper.class","value":"org.apache.hadoop.mapred.lib.RegexMapper"},{"name":"mapred.combiner.class","value":"org.apache.hadoop.mapred.lib.LongSumReducer"},{"name":"mapred.reducer.class","value":"org.apache.hadoop.mapred.lib.LongSumReducer"},{"name":"mapred.output.key.class","value":"org.apache.hadoop.io.Text"},{"name":"mapred.output.value.class","value":"org.apache.hadoop.io.LongWritable"}]</field>
+    <field type="CharField" name="jar_path">/user/hue/jobsub/examples/hadoop-examples.jar</field>
+  </object>
+  <object pk="3" model="jobsub.ooziemapreduceaction">
+    <field type="CharField" name="files">[]</field>
+    <field type="CharField" name="archives">[]</field>
+    <field type="CharField" name="job_properties">[{"name":"org.apache.hadoop.io.Text","value":"$num_maps"},{"name":"mapred.reduce.tasks","value":"$num_reduces"},{"name":"mapred.mapper.class","value":"org.apache.hadoop.examples.SleepJob"},{"name":"mapred.reducer.class","value":"org.apache.hadoop.examples.SleepJob"},{"name":"mapred.mapoutput.key.class","value":"org.apache.hadoop.io.IntWritable"},{"name":"mapred.mapoutput.value.class","value":"org.apache.hadoop.io.NullWritable"},{"name":"mapred.output.format.class","value":"org.apache.hadoop.mapred.lib.NullOutputFormat"},{"name":"mapred.input.format.class","value":"org.apache.hadoop.examples.SleepJob$$SleepInputFormat"},{"name":"mapred.partitioner.class","value":"org.apache.hadoop.examples.SleepJob"},{"name":"mapred.speculative.execution","value":"false"},{"name":"sleep.job.map.sleep.time","value":"$map_sleep_time"},{"name":"sleep.job.reduce.sleep.time","value":"$reduce_sleep_time"}]</field>
+    <field type="CharField" name="jar_path">/user/hue/jobsub/examples/hadoop-examples.jar</field>
+  </object>
+  <object pk="1" model="jobsub.ooziestreamingaction">
+    <field type="CharField" name="files">["/user/hue/jobsub/examples/wordcount.py"]</field>
+    <field type="CharField" name="archives">[]</field>
+    <field type="CharField" name="job_properties">[{"name":"mapred.input.dir","value":"/user/hue/jobsub/sample_data"},{"name":"mapred.output.dir","value":"$output_directory"},{"name":"mapred.reduce.tasks","value":"1"}]</field>
+    <field type="CharField" name="mapper">python wordcount.py map</field>
+    <field type="CharField" name="reducer">python wordcount.py reduce</field>
   </object>
 </django-objects>

+ 55 - 19
apps/jobsub/src/jobsub/management/commands/jobsub_setup.py

@@ -34,6 +34,11 @@ import jobsub.conf
 
 LOG = logging.getLogger(__name__)
 
+
+# The setup_level value for the CheckForSetup table
+JOBSUB_SETUP_LEVEL = 200        # Stands for Hue 2.0.0
+
+
 class Command(NoArgsCommand):
   """Creates file system for testing."""
   def handle_noargs(self, **options):
@@ -42,8 +47,8 @@ class Command(NoArgsCommand):
       remote_fs.setuser(remote_fs.superuser)
     LOG.info("Using remote fs: %s" % str(remote_fs))
 
-    # Copy over examples/ and script_templates/ directories
-    for dirname in ("examples", "script_templates"):
+    # Copy over examples/
+    for dirname in ("examples",):
       local_dir = os.path.join(jobsub.conf.LOCAL_DATA_DIR.get(), dirname)
       remote_dir = posixpath.join(jobsub.conf.REMOTE_DATA_DIR.get(), dirname)
       copy_dir(local_dir, remote_fs, remote_dir)
@@ -53,25 +58,30 @@ class Command(NoArgsCommand):
       remote_fs,
       posixpath.join(jobsub.conf.REMOTE_DATA_DIR.get(), "sample_data"))
 
-    # Also copy over Hadoop examples and streaming jars
-    local_src = hadoop.conf.HADOOP_EXAMPLES_JAR.get()
-    if local_src is None:
-      raise Exception('Failed to locate the Hadoop example jar')
-    remote_dst = posixpath.join(jobsub.conf.REMOTE_DATA_DIR.get(), "examples", "hadoop-examples.jar")
-    copy_file(local_src, remote_fs, remote_dst)
-
     # Write out the models too
     fixture_path = os.path.join(os.path.dirname(__file__), "..", "..", "fixtures", "example_data.xml")
     examples = django.core.serializers.deserialize("xml", open(fixture_path))
     sample_user = None
-    sample_job_designs = []
+    sample_oozie_designs = []
+    sample_oozie_abstract_actions = {}      # pk -> object
+    sample_oozie_concrete_actions = {}      # oozieaction_ptr_id -> object
+
     for example in examples:
       if isinstance(example.object, User):
         sample_user = example
-      elif isinstance(example.object, jobsub.models.JobDesign):
-        sample_job_designs.append(example)
+      elif isinstance(example.object, jobsub.models.OozieDesign):
+        sample_oozie_designs.append(example)
+      elif type(example.object) in (jobsub.models.OozieMapreduceAction,
+                                    jobsub.models.OozieJavaAction,
+                                    jobsub.models.OozieStreamingAction):
+        key = example.object.oozieaction_ptr_id
+        sample_oozie_concrete_actions[key] = example
+      elif type(example.object) is jobsub.models.OozieAction:
+        key = example.object.pk
+        sample_oozie_abstract_actions[key] = example
       else:
         raise Exception("Unexpected fixture type.")
+
     if sample_user is None:
       raise Exception("Expected sample user fixture.")
     # Create the sample user if it doesn't exist
@@ -81,11 +91,32 @@ class Command(NoArgsCommand):
       sample_user.object.pk = None
       sample_user.object.id = None
       sample_user.save()
-    for j in sample_job_designs:
-      j.object.id = None
-      j.object.pk = None
-      j.object.owner_id = sample_user.object.id
-      j.save()
+
+    # Create the designs
+    for d in sample_oozie_designs:
+      #
+      # OozieDesign          ----many-to-one--->  OozieAction
+      #
+      # OozieMapreduceAction -----one-to-one--->  OozieAction
+      # OozieStreamingAction -----one-to-one--->  OozieAction
+      # OozieJavaAction      -----one-to-one--->  OozieAction
+      #
+      # We find the OozieAction pk and link everything back together
+      #
+      abstract_action_id = d.object.root_action_id
+      abstract_action = sample_oozie_abstract_actions[abstract_action_id]
+      concrete_action = sample_oozie_concrete_actions[str(abstract_action_id)]
+
+      concrete_action.object.action_type = abstract_action.object.action_type
+      concrete_action.object.pk = None
+      concrete_action.object.id = None
+      concrete_action.object.save()
+
+      d.object.id = None
+      d.object.pk = None
+      d.object.owner_id = sample_user.object.id
+      d.object.root_action = concrete_action.object
+      d.object.save()
 
     # Upon success, write to the database
     try:
@@ -93,6 +124,7 @@ class Command(NoArgsCommand):
     except jobsub.models.CheckForSetup.DoesNotExist:
       entry = jobsub.models.CheckForSetup(id=1)
     entry.setup_run = True
+    entry.setup_level = JOBSUB_SETUP_LEVEL
     entry.save()
 
   def has_been_setup(self):
@@ -103,7 +135,8 @@ class Command(NoArgsCommand):
       entry = jobsub.models.CheckForSetup.objects.get(id=1)
     except jobsub.models.CheckForSetup.DoesNotExist:
       return False
-    return entry.setup_run
+    return entry.setup_run and entry.setup_level >= JOBSUB_SETUP_LEVEL
+
 
 def copy_dir(local_dir, remote_fs, remote_dir):
   # Hadoop mkdir is always recursive.
@@ -113,6 +146,9 @@ def copy_dir(local_dir, remote_fs, remote_dir):
     remote_dst = posixpath.join(remote_dir, f)
     copy_file(local_src, remote_fs, remote_dst)
 
+
+CHUNK_SIZE = 65536
+
 def copy_file(local_src, remote_fs, remote_dst):
   if remote_fs.exists(remote_dst):
     LOG.info("%s already exists.  Skipping." % remote_dst)
@@ -125,7 +161,7 @@ def copy_file(local_src, remote_fs, remote_dst):
     try:
       dst = remote_fs.open(remote_dst, "w")
       try:
-        shutil.copyfileobj(src, dst)
+        shutil.copyfileobj(src, dst, CHUNK_SIZE)
         LOG.info("Copied %s -> %s" % (local_src, remote_dst))
       finally:
         dst.close()

+ 26 - 0
apps/jobsub/src/jobsub/templates/list_designs.mako

@@ -36,6 +36,10 @@ ${layout.menubar(section='designs')}
             <a href="${ url('jobsub.views.new_design', action_type='mapreduce') }" class="btn">Create Mapreduce Design</a>
             <a href="${ url('jobsub.views.new_design', action_type='streaming') }" class="btn">Create Streaming Design</a>
             <a href="${ url('jobsub.views.new_design', action_type='java') }" class="btn">Create Java Design</a>
+            %if show_install_examples:
+              <a class="btn installSamplesLink" href="javascript:void(0)"
+                data-confirmation-url="${url('jobsub.views.setup')}">Install Samples</a>
+            %endif
         </p>
     </div>
 
@@ -116,6 +120,21 @@ ${layout.menubar(section='designs')}
 	</form>
 </div>
 
+
+<div id="installSamples" class="modal hide fade">
+	<form id="installSamplesForm" action="${url('jobsub.views.setup')}" method="POST">
+        <div class="modal-header">
+            <a href="#" class="close">&times;</a>
+            <h3>Install sample job designs?</h3>
+        </div>
+        <div class="modal-footer">
+            <input type="submit" class="btn primary" value="Yes"/>
+            <a href="#" class="btn secondary hideModal">No</a>
+        </div>
+	</form>
+</div>
+
+
 <script type="text/javascript" charset="utf-8">
     $(document).ready(function() {
         $(".modal").modal({
@@ -134,6 +153,13 @@ ${layout.menubar(section='designs')}
             $("#deleteWf").modal("hide");
         });
 
+        $(".installSamplesLink").click(function(){
+            $("#installSamples").modal("show");
+        });
+        $("#installSamples .hideModal").click(function(){
+            $("#installSamples").modal("hide");
+        });
+
         $(".submitConfirmation").click(function(){
             var _this = $(this);
             var _action = _this.attr("data-submit-url");

+ 1 - 1
apps/jobsub/src/jobsub/urls.py

@@ -35,5 +35,5 @@ urlpatterns = patterns(
   url(r'^list_history$', 'views.list_history'),
 
   # Setup
-  #url(r'^setup/$', 'views.setup'),
+  url(r'^setup/$', 'views.setup'),
 )

+ 12 - 0
apps/jobsub/src/jobsub/views.py

@@ -38,6 +38,7 @@ from desktop.lib.django_util import render, PopupException, extract_field_data
 from desktop.log.access import access_warn
 
 from jobsub import models, submit
+from jobsub.management.commands import jobsub_setup
 from jobsub.oozie_lib.oozie_api import get_oozie
 import jobsub.forms
 
@@ -135,12 +136,15 @@ def list_designs(request):
   if name:
     data = data.filter(name__icontains=name)
   data = data.order_by('-last_modified')
+  show_install_examples = \
+      request.user.is_superuser and not jobsub_setup.Command().has_been_setup()
 
   return render("list_designs.mako", request, {
     'designs': list(data),
     'currentuser':request.user,
     'owner': owner,
     'name': name,
+    'show_install_examples': show_install_examples,
   })
 
 
@@ -247,6 +251,14 @@ def submit_design(request, design_id):
   return redirect(urlresolvers.reverse(oozie_job, kwargs={'jobid': jobid}))
 
 
+def setup(request):
+  """Installs jobsub examples."""
+  if request.method != "POST":
+    raise PopupException('Please use a POST request to install the examples.')
+  jobsub_setup.Command().handle_noargs()
+  return redirect(urlresolvers.reverse(list_designs))
+
+
 
 # See http://wiki.apache.org/hadoop/JobConfFile
 _STD_PROPERTIES = [