#32 at this point we have the first flow example running with one job…

… in the cluster; next steps are adding more jobs, adding a flow runner for the EMR cluster, and fluentizing the Java/Sprin XML API
paulhoule · Oct 10, 2013 · c2d27d8 · c2d27d8
1 parent 738dc26
commit c2d27d8
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/haruhi/src/main/java/com/ontology2/haruhi/FlowApp.java b/haruhi/src/main/java/com/ontology2/haruhi/FlowApp.java
@@ -11,6 +11,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.PeekingIterator;
 import com.ontology2.centipede.shell.CommandLineApplication;
+import com.ontology2.haruhi.flows.Flow;
 
 public class FlowApp extends CommandLineApplication {
     private static Log logger = LogFactory.getLog(FlowApp.class);
@@ -45,8 +46,11 @@ protected void _run(String[] arguments) throws Exception {
             usage();
 
         String flowId=a.next();
+        Flow f=applicationContext.getBean(flowId,Flow.class);
+
         List<String> flowArgs=Lists.newArrayList();
         Iterators.addAll(flowArgs, a);
+        cluster.runFlow(jar, f, flowArgs);
     }
 
     private void usage() {

diff --git a/haruhi/src/main/java/com/ontology2/haruhi/LocalCmdCluster.java b/haruhi/src/main/java/com/ontology2/haruhi/LocalCmdCluster.java
@@ -73,12 +73,16 @@ public void setMavenRepoPath(String mavenRepoPath) {
 
     @Override
     public void runFlow(MavenManagedJar jar, Flow f,List<String> flowArgs) throws Exception {
-        List<FlowStep> steps=f.generateSteps(flowArgs);
-        if(steps instanceof JobStep) {
-            JobStep jobStep=(JobStep) steps;
+        for(FlowStep that:f.generateSteps(flowArgs))
+            runStep(jar, flowArgs, that);
+    }
+
+    private void runStep(MavenManagedJar jar, List<String> flowArgs, FlowStep that) throws Exception {
+        if(that instanceof JobStep) {
+            JobStep jobStep=(JobStep) that;
             this.runJob(jar, jobStep.getStepArgs(flowArgs));
         } else {
-            throw new RuntimeException("Could not process step of type "+steps.getClass());
+            throw new RuntimeException("Could not process step of type "+that.getClass());
         }
     }
 };
diff --git a/haruhi/src/main/resources/com/ontology2/haruhi/shell/applicationContext.xml b/haruhi/src/main/resources/com/ontology2/haruhi/shell/applicationContext.xml
@@ -54,6 +54,7 @@
             <bean class="com.ontology2.haruhi.flows.JobStep">
                 <constructor-arg>
                   <list>
+                     <value>'run'</value>
                      <value>'freebaseRDFPrefilter'</value>
                      <value>pos[0]+'freebase-rdf-'+pos[1]+'/'</value>
                      <value>tmpDir+'preprocessed/'+pos[1]+'/'</value>