Merge pull request #18 from justinliucs/master

Merge from base project.

pom.xml

@@ -76,6 +76,11 @@
 			<groupId>org.apache.hive</groupId>
 			<artifactId>hive-jdbc</artifactId>
 			<version>0.9.0</version>
+		</dependency>
+		<dependency>
+			<groupId>com.googlecode.json-simple</groupId>
+			<artifactId>json-simple</artifactId>
+			<version>1.1</version>
 		</dependency>
 	</dependencies>
 

src/main/java/haflow/engine/oozie/EndModuleGenerator.java

@@ -2,6 +2,7 @@ package haflow.engine.oozie;
 
 import haflow.dto.entity.Node;
 
+import java.util.List;
 import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -13,7 +14,7 @@ public class EndModuleGenerator extends OozieXmlGenerator {
 
 	@Override
 	public Document generate(Map<String, String> configurations,
-			Map<String, Node> inputs, Map<String, Node> outputs) {
+			Map<String, Node> inputs, Map<String, Node> outputs, List<String> arguments) {
 		try {
 			String name = configurations.get("name");
 			String xml = "<end name=\"" + name + "\"/>";

src/main/java/haflow/engine/oozie/HiveModuleGenerator.java

@@ -2,6 +2,7 @@ package haflow.engine.oozie;
 
 import haflow.dto.entity.Node;
 
+import java.util.List;
 import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -23,7 +24,7 @@ public class HiveModuleGenerator extends OozieXmlGenerator {
 
 	@Override
 	public Document generate(Map<String, String> configurations,
-			Map<String, Node> inputs, Map<String, Node> outputs) {
+			Map<String, Node> inputs, Map<String, Node> outputs, List<String> arguments) {
 		try {
 			String name = configurations.get("name");
 

src/main/java/haflow/engine/oozie/JavaModuleGenerator.java

@@ -2,6 +2,7 @@ package haflow.engine.oozie;
 
 import haflow.dto.entity.Node;
 
+import java.util.List;
 import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -26,7 +27,7 @@ public class JavaModuleGenerator extends OozieXmlGenerator {
 	
 	@Override
 	public Document generate(Map<String, String> configurations,
-			Map<String, Node> inputs, Map<String, Node> outputs) {
+			Map<String, Node> inputs, Map<String, Node> outputs, List<String> arguments) {
 		try {
 			String name = configurations.get("name");
 
@@ -36,12 +37,19 @@ public class JavaModuleGenerator extends OozieXmlGenerator {
 			
 			String main_class = configurations.get(MAIN_CLASS);
 //			String java_opt = configurations.get(JAVA_OPT);
-			String argument = configurations.get(ARG);
-			System.out.println("argument: " + argument);
-			
+//			String argument = configurations.get(ARG);
+//			System.out.println("argument: " + argument);
+						
 			String ok = outputs.get("ok").getName();
 			String error = outputs.get("error").getName();
 			
+			StringBuilder sb = new StringBuilder();
+			if( arguments != null){
+				for(int i = 0; i < arguments.size(); i++){
+					sb.append("<arg>" + arguments.get(i) + "</arg>" + "\n");
+				}
+			}
+			
 			String xml = "<action name=\"" + name + "\">" + "\n" + "<java>"
 					+ "\n" + "<job-tracker>" + job_tracker + "</job-tracker>" + "\n"
 					+ "<name-node>" + name_node + "</name-node>" + "\n"
@@ -50,7 +58,8 @@ public class JavaModuleGenerator extends OozieXmlGenerator {
 					+ "<value>" + queue_name + "</value>" + "\n" + "</property>"
 					+ "\n" + "</configuration>" + "\n" + "<main-class>"
 					+ main_class + "</main-class>" + "\n"
-					+ "<arg>" + argument + "</arg>" + "\n" + "</java>"
+					+ sb.toString()
+					+ "</java>"
 					+ "\n" + "<ok to=\"" + ok + "\"/>" + "\n"//ok
 					+ "<error to=\"" +error + "\"/>" + "\n" + "</action>";
 			return DocumentBuilderFactory.newInstance().newDocumentBuilder()

src/main/java/haflow/engine/oozie/KillModuleGenerator.java

@@ -2,6 +2,7 @@ package haflow.engine.oozie;
 
 import haflow.dto.entity.Node;
 
+import java.util.List;
 import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -13,7 +14,7 @@ public class KillModuleGenerator extends OozieXmlGenerator {
 
 	@Override
 	public Document generate(Map<String, String> configurations,
-			Map<String, Node> inputs, Map<String, Node> outputs) {
+			Map<String, Node> inputs, Map<String, Node> outputs, List<String> arguments) {
 		try {
 			String name = configurations.get("name");
 			String xml = "<kill name=\""

src/main/java/haflow/engine/oozie/OozieEngine.java

@@ -166,46 +166,46 @@ public class OozieEngine extends AbstractEngine {
 					boolean deloyedLocally = this.getFlowDeployService()
 							.deployFlowLocal(localDeployPath, workflowXml,
 									getJarPaths(nodes, moduleClasses));
-					// if (deloyedLocally) {
-					// messageBuilder.append(flowName
-					// + " has been deployed locally!" + "\n");
-					//
-					// String hdfsDeployPath = this.getClusterConfiguration()
-					// .getProperty(
-					// ClusterConfiguration.WORKSPACE_HDFS)
-					// + flowName;
-					// boolean deleted = this.getHdfsService()
-					// .deleteDirectory(hdfsDeployPath);
-					// if (deleted) {
-					// messageBuilder.append("Old folder deleted: "
-					// + hdfsDeployPath + "\n");
-					// }
-					//
-					// boolean deployedToHdfs = this.getHdfsService()
-					// .uploadFile(localDeployPath, hdfsDeployPath);
-					// if (deployedToHdfs) {
-					// messageBuilder.append(flowName
-					// + " has been uploaded to hdfs!" + "\n");
-					//
-					// String jobId = this.getOozieService().runJob(
-					// flowName);
-					// if (jobId == null) {
-					// messageBuilder.append("Failed to commit job: "
-					// + flowName + "\n");
-					// } else {
-					// messageBuilder.append("Job commited! Job id : "
-					// + jobId + "\n");
-					// model.setCommitted(true);
-					// model.setJobId(jobId);
-					// }
-					// } else {
-					// messageBuilder.append(flowName
-					// + " failed to be uploaded to hdfs!" + "\n");
-					// }
-					// } else {
-					// messageBuilder.append(flowName
-					// + " failed to be deployed locally!" + "\n");
-					// }
+					if (deloyedLocally) {
+						messageBuilder.append(flowName
+								+ " has been deployed locally!" + "\n");
+
+						String hdfsDeployPath = this.getClusterConfiguration()
+								.getProperty(
+										ClusterConfiguration.WORKSPACE_HDFS)
+								+ flowName;
+						boolean deleted = this.getHdfsService()
+								.deleteDirectory(hdfsDeployPath);
+						if (deleted) {
+							messageBuilder.append("Old folder deleted: "
+									+ hdfsDeployPath + "\n");
+						}
+
+						boolean deployedToHdfs = this.getHdfsService()
+								.uploadFile(localDeployPath, hdfsDeployPath);
+						if (deployedToHdfs) {
+							messageBuilder.append(flowName
+									+ " has been uploaded to hdfs!" + "\n");
+
+							String jobId = this.getOozieService().runJob(
+									flowName);
+							if (jobId == null) {
+								messageBuilder.append("Failed to commit job: "
+										+ flowName + "\n");
+							} else {
+								messageBuilder.append("Job commited! Job id : "
+										+ jobId + "\n");
+								model.setCommitted(true);
+								model.setJobId(jobId);
+							}
+						} else {
+							messageBuilder.append(flowName
+									+ " failed to be uploaded to hdfs!" + "\n");
+						}
+					} else {
+						messageBuilder.append(flowName
+								+ " failed to be deployed locally!" + "\n");
+					}
 				}
 			}
 
@@ -306,7 +306,7 @@ public class OozieEngine extends AbstractEngine {
 			}
 
 			Module moduleProtype = moduleClass.getAnnotation(Module.class);
-
+			List<String> arguments = null;
 			OozieXmlGenerator gen = null;
 			switch (moduleProtype.type()) {
 			case START:
@@ -322,8 +322,9 @@ public class OozieEngine extends AbstractEngine {
 				AbstractJavaModule moduleInstance = (AbstractJavaModule) moduleClass
 						.newInstance();
 				configurations.put("main_class", moduleInstance.getMainClass());
-				configurations.put("arg",
-						moduleInstance.getArguments(userConfs));
+//				configurations.put("arg",
+//						moduleInstance.getArguments(userConfs));
+				arguments = moduleInstance.getArguments(userConfs);
 				gen = new JavaModuleGenerator();
 				break;
 			case HIVE:
@@ -346,7 +347,7 @@ public class OozieEngine extends AbstractEngine {
 				break;
 			}
 			if (gen != null) {
-				Document doc = gen.generate(configurations, null, outputs);
+				Document doc = gen.generate(configurations, null, outputs, arguments);
 				TransformerFactory transFactory = TransformerFactory
 						.newInstance();
 				Transformer transformer;

src/main/java/haflow/engine/oozie/OozieService.java

@@ -47,7 +47,7 @@ public class OozieService {
 		Properties conf = wc.createConfiguration();
 		// set the workflow application path
 		conf.setProperty(OozieClient.APP_PATH,
-				"hdfs://m150:9000/user/root/examples/apps/z-java-module");
+				"hdfs://m150:9000/user/root/examples/apps/race_random_forest2");//z-java-module
 
 		// setting workflow parameters
 		conf.setProperty("nameNode", "hdfs://m150:9000");

src/main/java/haflow/engine/oozie/OozieXmlGenerator.java

@@ -2,11 +2,12 @@ package haflow.engine.oozie;
 
 import haflow.dto.entity.Node;
 
+import java.util.List;
 import java.util.Map;
 
 import org.w3c.dom.Document;
 
 public abstract class OozieXmlGenerator {
 	public abstract Document generate(Map<String, String> configurations,
-			Map<String, Node> inputs, Map<String, Node> outputs);
+			Map<String, Node> inputs, Map<String, Node> outputs, List<String> arguments);
 }

src/main/java/haflow/engine/oozie/StartModuleGenerator.java

@@ -2,6 +2,7 @@ package haflow.engine.oozie;
 
 import haflow.dto.entity.Node;
 
+import java.util.List;
 import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -13,7 +14,7 @@ public class StartModuleGenerator extends OozieXmlGenerator {
 
 	@Override
 	public Document generate(Map<String, String> configurations,
-			Map<String, Node> inputs, Map<String, Node> outputs) {
+			Map<String, Node> inputs, Map<String, Node> outputs, List<String> arguments) {
 		System.out.println(outputs.size());
 		try {
 			String xml = "<start to=\"" + outputs.get("to").getName() + "\"/>";

src/main/java/haflow/module/AbstractJavaModule.java

@@ -1,5 +1,6 @@
 package haflow.module;
 
+import java.util.List;
 import java.util.Map;
 
 
@@ -7,11 +8,12 @@ public abstract class AbstractJavaModule extends AbstractModule{
 	
 	public abstract String getMainClass();
 	
-	public String getArguments(Map<String, String> configurations){
-		StringBuilder sb = new StringBuilder();
-		for(String key : configurations.keySet()){
-			sb.append("--" + key + " \"" + configurations.get(key) + "\" ");
-		}
-		return sb.toString();
+	public List<String> getArguments(Map<String, String> configurations){
+//		StringBuilder sb = new StringBuilder();
+//		for(String key : configurations.keySet()){
+//			sb.append("--" + key + " \"" + configurations.get(key) + "\" ");
+//		}
+//		return sb.toString();
+		return null;
 	}
 }

src/main/java/haflow/module/datamining/BuildForestModule.java

@@ -8,9 +8,11 @@ import haflow.module.ModuleConfigurationType;
 import haflow.module.ModuleEndpoint;
 import haflow.module.ModuleType;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
-@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bdac", name = "BuildForest", category = "datamining", type = ModuleType.JAVA, 
+@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bdac", name = "BuildForest", category = "DataMining-Mahout", type = ModuleType.JAVA, 
 	configurations = {
 		@ModuleConfiguration(key = "data", displayName = "data: Data path", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
 		@ModuleConfiguration(key = "dataset", displayName = "dataset: Dataset path",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
@@ -40,23 +42,26 @@ public class BuildForestModule extends AbstractJavaModule {
 	}
 
 	@Override
-	public String getArguments(Map<String, String> configurations) {
-		Module module= BuildForestModule.class.getAnnotation(Module.class);
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
 		ModuleConfiguration[] confs = module.configurations();
 		
-		StringBuilder sb = new StringBuilder();
+		List<String> result = new ArrayList<String>();
 		for (String key : configurations.keySet()) {
 			ModuleConfigurationType confType = getConfigurationType(key, confs);
 			switch(confType){
 			case BOOLEAN:
 				String boolValue = configurations.get(key);
-				if( boolValue.equals("true"))
-					sb.append("--" + key + " ");
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
 				break;
 			case PLAIN_TEXT:
 				String textValue = configurations.get(key).trim();
-				if( textValue.length() > 0)
-					sb.append("--" + key + " \"" + configurations.get(key) + "\" ");
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
 				break;
 			case OTHER:				
 			default:
@@ -64,7 +69,7 @@ public class BuildForestModule extends AbstractJavaModule {
 				break;
 			}
 		}
-		return sb.toString();
+		return result;
 	}
 	
 	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){

src/main/java/haflow/module/datamining/CanopyClusterModule.java

@@ -0,0 +1,83 @@
+package haflow.module.datamining;
+
+import haflow.module.AbstractJavaModule;
+import haflow.module.DataType;
+import haflow.module.Module;
+import haflow.module.ModuleConfiguration;
+import haflow.module.ModuleConfigurationType;
+import haflow.module.ModuleEndpoint;
+import haflow.module.ModuleType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bd00", name = "CanopyCluster", category = "DataMining-Mahout", type = ModuleType.JAVA, 
+	configurations = {
+		@ModuleConfiguration(key = "input", displayName = "input", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "output", displayName = "output",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "distanceMeasure", displayName = "distanceMeasure", pattern = "^(.*)$",type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "t1", displayName = "t1", pattern = "^(.*)$",type=ModuleConfigurationType.PLAIN_TEXT), 
+		@ModuleConfiguration(key = "t2", displayName = "t2", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "t3", displayName = "t3", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "t4", displayName = "t4", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "clusterFilter", displayName = "clusterFilter", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "overwrite", displayName = "overwrite", pattern = "^(.*)$", type=ModuleConfigurationType.BOOLEAN),
+		@ModuleConfiguration(key = "outlierThreshold", displayName = "outlierThreshold", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT)},
+	inputs = { @ModuleEndpoint(name = "from", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) }, outputs = {
+		@ModuleEndpoint(name = "ok", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText),
+		@ModuleEndpoint(name = "error", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) })
+public class CanopyClusterModule extends AbstractJavaModule {
+
+	@Override
+	public boolean validate(Map<String, String> configurations,
+			Map<String, String> inputs, Map<String, String> outputs) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	@Override
+	public String getMainClass() {
+		return "org.apache.mahout.clustering.canopy.CanopyDriver";
+	}
+
+	@Override
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
+		ModuleConfiguration[] confs = module.configurations();
+		
+		List<String> result = new ArrayList<String>();
+		for (String key : configurations.keySet()) {
+			ModuleConfigurationType confType = getConfigurationType(key, confs);
+			switch(confType){
+			case BOOLEAN:
+				String boolValue = configurations.get(key);
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
+				break;
+			case PLAIN_TEXT:
+				String textValue = configurations.get(key).trim();
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
+				break;
+			case OTHER:				
+			default:
+				System.out.println("Invalid Parameters!");
+				break;
+			}
+		}
+		return result;
+	}
+	
+	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){
+		for( ModuleConfiguration conf : confs){
+			if( key.equals(conf.key()))
+				return conf.type();
+		}
+		return ModuleConfigurationType.OTHER;
+	}
+
+}

src/main/java/haflow/module/datamining/DescribeModule.java

@@ -8,9 +8,11 @@ import haflow.module.ModuleConfigurationType;
 import haflow.module.ModuleEndpoint;
 import haflow.module.ModuleType;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
-@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bdab", name = "Describe", category = "datamining", type = ModuleType.JAVA, configurations = {
+@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bdab", name = "Describe", category = "DataMining-Mahout", type = ModuleType.JAVA, configurations = {
 		@ModuleConfiguration(key = "path", displayName = "path: Data path", pattern = "^(.*)$", type = ModuleConfigurationType.PLAIN_TEXT),
 		@ModuleConfiguration(key = "descriptor", displayName = "descriptor: Data descriptor", pattern = "^(.*)$", type = ModuleConfigurationType.PLAIN_TEXT),
 		@ModuleConfiguration(key = "file", displayName = "file: Path to generated descriptor file", pattern = "^(.*)$", type = ModuleConfigurationType.PLAIN_TEXT),
@@ -32,23 +34,34 @@ public class DescribeModule extends AbstractJavaModule {
 	}
 
 	@Override
-	public String getArguments(Map<String, String> configurations) {
-		Module module = DescribeModule.class.getAnnotation(Module.class);
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module = this.getClass().getAnnotation(Module.class);
 		ModuleConfiguration[] confs = module.configurations();
 
-		StringBuilder sb = new StringBuilder();
+		List<String> result = new ArrayList<String>();
 		for (String key : configurations.keySet()) {
 			ModuleConfigurationType confType = getConfigurationType(key, confs);
 			switch (confType) {
 			case BOOLEAN:
 				String boolValue = configurations.get(key);
-				if( boolValue.equals("true"))
-					sb.append("--" + key + " ");
+				if( boolValue.equals("true")){
+					result.add("--" + key);
+				}
 				break;
 			case PLAIN_TEXT:
 				String textValue = configurations.get(key).trim();
-				if( textValue.length() > 0)
-					sb.append("--" + key + " \"" + configurations.get(key) + "\" ");
+				if( textValue.length() > 0){
+					if(key.equals("descriptor")){
+						result.add("--" + key);
+						String[] descriptors = textValue.split(" ");
+						for(String desp : descriptors){
+							result.add(desp);
+						}
+					}else{
+						result.add("--" + key);
+						result.add(configurations.get(key));
+					}
+				}
 				break;
 			case OTHER:
 			default:
@@ -56,7 +69,7 @@ public class DescribeModule extends AbstractJavaModule {
 				break;
 			}
 		}
-		return sb.toString();
+		return result;
 	}
 
 	private ModuleConfigurationType getConfigurationType(String key,

src/main/java/haflow/module/datamining/TestForestModule.java

@@ -8,9 +8,11 @@ import haflow.module.ModuleConfigurationType;
 import haflow.module.ModuleEndpoint;
 import haflow.module.ModuleType;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
-@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bdad", name = "TestForest", category = "datamining", type = ModuleType.JAVA, 
+@Module(id = "ada600a8-aa63-968a-ca46-4356a0e0bdad", name = "TestForest", category = "DataMining-Mahout", type = ModuleType.JAVA, 
 	configurations = {
 		@ModuleConfiguration(key = "input", displayName = "input : Path to job input directory.", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
 		@ModuleConfiguration(key = "dataset", displayName = "dataset : Dataset path",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
@@ -36,24 +38,26 @@ public class TestForestModule extends AbstractJavaModule {
 	}
 
 	@Override
-	public String getArguments(Map<String, String> configurations) {
-		Module module= TestForestModule.class.getAnnotation(Module.class);
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
 		ModuleConfiguration[] confs = module.configurations();
 		
-		StringBuilder sb = new StringBuilder();
+		List<String> result = new ArrayList<String>();
 		for (String key : configurations.keySet()) {
 			ModuleConfigurationType confType = getConfigurationType(key, confs);
 			switch(confType){
 			case BOOLEAN:
 				String boolValue = configurations.get(key);
-				boolValue = boolValue.trim().toLowerCase();
-				if( boolValue.equals("true") || boolValue.equals("1"))
-					sb.append("--" + key + " ");
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
 				break;
 			case PLAIN_TEXT:
 				String textValue = configurations.get(key).trim();
-				if( textValue.length() > 0)
-					sb.append("--" + key + " \"" + configurations.get(key) + "\" ");
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
 				break;
 			case OTHER:				
 			default:
@@ -61,7 +65,7 @@ public class TestForestModule extends AbstractJavaModule {
 				break;
 			}
 		}
-		return sb.toString();
+		return result;
 	}
 	
 	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){

src/main/java/haflow/module/dm/weka/WekaAssociationModule.java

@@ -0,0 +1,75 @@
+package haflow.module.dm.weka;
+
+import haflow.module.AbstractJavaModule;
+import haflow.module.DataType;
+import haflow.module.Module;
+import haflow.module.ModuleConfiguration;
+import haflow.module.ModuleConfigurationType;
+import haflow.module.ModuleEndpoint;
+import haflow.module.ModuleType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Module(id = "ada600a8-aa63-968a-ca46-0156a0e0bdff", name = "Apriori", category = "DataMining-Weka", type = ModuleType.JAVA, 
+	configurations = {
+		@ModuleConfiguration(key = "input", displayName = "input", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "output", displayName = "output",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT)},
+	inputs = { @ModuleEndpoint(name = "from", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) }, outputs = {
+		@ModuleEndpoint(name = "ok", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText),
+		@ModuleEndpoint(name = "error", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) })
+public class WekaAssociationModule extends AbstractJavaModule {
+
+	@Override
+	public boolean validate(Map<String, String> configurations,
+			Map<String, String> inputs, Map<String, String> outputs) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	@Override
+	public String getMainClass() {
+		return "org.apache.mahout.clustering.canopy.CanopyDriver";
+	}
+
+	@Override
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
+		ModuleConfiguration[] confs = module.configurations();
+		
+		List<String> result = new ArrayList<String>();
+		for (String key : configurations.keySet()) {
+			ModuleConfigurationType confType = getConfigurationType(key, confs);
+			switch(confType){
+			case BOOLEAN:
+				String boolValue = configurations.get(key);
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
+				break;
+			case PLAIN_TEXT:
+				String textValue = configurations.get(key).trim();
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
+				break;
+			case OTHER:				
+			default:
+				System.out.println("Invalid Parameters!");
+				break;
+			}
+		}
+		return result;
+	}
+	
+	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){
+		for( ModuleConfiguration conf : confs){
+			if( key.equals(conf.key()))
+				return conf.type();
+		}
+		return ModuleConfigurationType.OTHER;
+	}
+
+}

src/main/java/haflow/module/dm/weka/WekaClassifierModule.java

@@ -0,0 +1,75 @@
+package haflow.module.dm.weka;
+
+import haflow.module.AbstractJavaModule;
+import haflow.module.DataType;
+import haflow.module.Module;
+import haflow.module.ModuleConfiguration;
+import haflow.module.ModuleConfigurationType;
+import haflow.module.ModuleEndpoint;
+import haflow.module.ModuleType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Module(id = "ada600a8-aa63-968a-ca46-0056a0e0bdff", name = "KMeans", category = "DataMining-Weka", type = ModuleType.JAVA, 
+	configurations = {
+		@ModuleConfiguration(key = "input", displayName = "input", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "output", displayName = "output",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT)},
+	inputs = { @ModuleEndpoint(name = "from", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) }, outputs = {
+		@ModuleEndpoint(name = "ok", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText),
+		@ModuleEndpoint(name = "error", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) })
+public class WekaClassifierModule extends AbstractJavaModule {
+
+	@Override
+	public boolean validate(Map<String, String> configurations,
+			Map<String, String> inputs, Map<String, String> outputs) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	@Override
+	public String getMainClass() {
+		return "org.apache.mahout.clustering.canopy.CanopyDriver";
+	}
+
+	@Override
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
+		ModuleConfiguration[] confs = module.configurations();
+		
+		List<String> result = new ArrayList<String>();
+		for (String key : configurations.keySet()) {
+			ModuleConfigurationType confType = getConfigurationType(key, confs);
+			switch(confType){
+			case BOOLEAN:
+				String boolValue = configurations.get(key);
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
+				break;
+			case PLAIN_TEXT:
+				String textValue = configurations.get(key).trim();
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
+				break;
+			case OTHER:				
+			default:
+				System.out.println("Invalid Parameters!");
+				break;
+			}
+		}
+		return result;
+	}
+	
+	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){
+		for( ModuleConfiguration conf : confs){
+			if( key.equals(conf.key()))
+				return conf.type();
+		}
+		return ModuleConfigurationType.OTHER;
+	}
+
+}

src/main/java/haflow/module/dm/weka/WekaClusterModule.java

@@ -0,0 +1,75 @@
+package haflow.module.dm.weka;
+
+import haflow.module.AbstractJavaModule;
+import haflow.module.DataType;
+import haflow.module.Module;
+import haflow.module.ModuleConfiguration;
+import haflow.module.ModuleConfigurationType;
+import haflow.module.ModuleEndpoint;
+import haflow.module.ModuleType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Module(id = "ada600a8-aa63-968a-ca46-0256a0e0bdff", name = "ID3", category = "DataMining-Weka", type = ModuleType.JAVA, 
+	configurations = {
+		@ModuleConfiguration(key = "input", displayName = "input", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "output", displayName = "output",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT)},
+	inputs = { @ModuleEndpoint(name = "from", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) }, outputs = {
+		@ModuleEndpoint(name = "ok", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText),
+		@ModuleEndpoint(name = "error", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) })
+public class WekaClusterModule extends AbstractJavaModule {
+
+	@Override
+	public boolean validate(Map<String, String> configurations,
+			Map<String, String> inputs, Map<String, String> outputs) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	@Override
+	public String getMainClass() {
+		return "org.apache.mahout.clustering.canopy.CanopyDriver";
+	}
+
+	@Override
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
+		ModuleConfiguration[] confs = module.configurations();
+		
+		List<String> result = new ArrayList<String>();
+		for (String key : configurations.keySet()) {
+			ModuleConfigurationType confType = getConfigurationType(key, confs);
+			switch(confType){
+			case BOOLEAN:
+				String boolValue = configurations.get(key);
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
+				break;
+			case PLAIN_TEXT:
+				String textValue = configurations.get(key).trim();
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
+				break;
+			case OTHER:				
+			default:
+				System.out.println("Invalid Parameters!");
+				break;
+			}
+		}
+		return result;
+	}
+	
+	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){
+		for( ModuleConfiguration conf : confs){
+			if( key.equals(conf.key()))
+				return conf.type();
+		}
+		return ModuleConfigurationType.OTHER;
+	}
+
+}

src/main/java/haflow/module/statistics/R/RSVDModule.java

@@ -0,0 +1,75 @@
+package haflow.module.statistics.R;
+
+import haflow.module.AbstractJavaModule;
+import haflow.module.DataType;
+import haflow.module.Module;
+import haflow.module.ModuleConfiguration;
+import haflow.module.ModuleConfigurationType;
+import haflow.module.ModuleEndpoint;
+import haflow.module.ModuleType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Module(id = "ada600a8-aa63-968a-ca46-ff00a0e0bdff", name = "SVD", category = "Statistics-R", type = ModuleType.JAVA, 
+	configurations = {
+		@ModuleConfiguration(key = "input", displayName = "input", pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT),
+		@ModuleConfiguration(key = "output", displayName = "output",pattern = "^(.*)$", type=ModuleConfigurationType.PLAIN_TEXT)},
+	inputs = { @ModuleEndpoint(name = "from", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) }, outputs = {
+		@ModuleEndpoint(name = "ok", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText),
+		@ModuleEndpoint(name = "error", minNumber = 1, maxNumber = 1, dataType = DataType.PlainText) })
+public class RSVDModule extends AbstractJavaModule {
+
+	@Override
+	public boolean validate(Map<String, String> configurations,
+			Map<String, String> inputs, Map<String, String> outputs) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+
+	@Override
+	public String getMainClass() {
+		return "org.apache.mahout.clustering.canopy.CanopyDriver";
+	}
+
+	@Override
+	public List<String> getArguments(Map<String, String> configurations) {
+		Module module= this.getClass().getAnnotation(Module.class);
+		ModuleConfiguration[] confs = module.configurations();
+		
+		List<String> result = new ArrayList<String>();
+		for (String key : configurations.keySet()) {
+			ModuleConfigurationType confType = getConfigurationType(key, confs);
+			switch(confType){
+			case BOOLEAN:
+				String boolValue = configurations.get(key);
+				if( boolValue.equals("true")){
+					result.add("--" + key );
+				}
+				break;
+			case PLAIN_TEXT:
+				String textValue = configurations.get(key).trim();
+				if( textValue.length() > 0){
+					result.add("--" + key);
+					result.add(configurations.get(key));
+				}
+				break;
+			case OTHER:				
+			default:
+				System.out.println("Invalid Parameters!");
+				break;
+			}
+		}
+		return result;
+	}
+	
+	private ModuleConfigurationType getConfigurationType(String key, ModuleConfiguration[] confs){
+		for( ModuleConfiguration conf : confs){
+			if( key.equals(conf.key()))
+				return conf.type();
+		}
+		return ModuleConfigurationType.OTHER;
+	}
+
+}

src/main/webapp/WEB-INF/views/logon.jsp

@@ -16,23 +16,24 @@
 <body>
 	<div class="wrapper">
 		<div class="title">
-			<h1>基于Hadoop的大数据分析平台</h1>
+			<h1>大数据分析服务平台</h1>
 		</div>
 		<div class="content">
 			<div class="left">
 				<img src="<%=basePath%>/images/logon.png" />
 			</div>
 			<div class="right">
-				<h2>登录</h2>
+				<!--  <h2>登录</h2> -->
 				<div class="field">
-					<span>用户名：</span><input type="text" />
-				</div>
+					<span style="float: left; width: 80px;">用户名：</span><input type="text" />
+				</div><br/>
 				<div class="field">
-					<span>密码：</span><input type="password" />
-				</div>
+					<span style="float: left; width: 80px;">密码：</span><input type="password" />
+				</div><br/>
 				<div>
-					<button type="button" onclick="window.location.href='main'">登录</button>
-					<button type="button" onclick="window.location.href='admin'">管理员登录</button>
+					<button style="width: 70px;height: 35px; font-family:Arial;font-size:14px;font-weight:bold;color:#333333;" type="button" onclick="window.location.href='main'">登录</button>
+					&nbsp;&nbsp;
+					<button style="width: 120px;height: 35px; font-family:Arial;font-size:14px;font-weight:normal;color:#333333;" type="button" onclick="window.location.href='admin'">管理员登录</button>
 				</div>
 			</div>
 			<div class="clear"></div>