aurpple / sparkjobserverclient Goto Github PK

import java.io.File;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.bluebreezecf.tools.sparkjobserver.api.ISparkJobServerClient;
import com.bluebreezecf.tools.sparkjobserver.api.ISparkJobServerClientConstants;
import com.bluebreezecf.tools.sparkjobserver.api.SparkJobConfig;
import com.bluebreezecf.tools.sparkjobserver.api.SparkJobInfo;
import com.bluebreezecf.tools.sparkjobserver.api.SparkJobJarInfo;
import com.bluebreezecf.tools.sparkjobserver.api.SparkJobResult;
import com.bluebreezecf.tools.sparkjobserver.api.SparkJobServerClientException;
import com.bluebreezecf.tools.sparkjobserver.api.SparkJobServerClientFactory;

/**
 * A sample shows how to use spark-job-server-client.
 * 
 * @author bluebreezecf
 * @since 2014-09-16
 *
 */
public class SparkJobServerClientTest {
	
	public static void main(String[] args) {
		ISparkJobServerClient client = null;
		try {
			client = SparkJobServerClientFactory.getInstance().createSparkJobServerClient("http://localhost:8090/");
			//GET /jars
			List<SparkJobJarInfo> jarInfos = client.getJars();
			for (SparkJobJarInfo jarInfo: jarInfos) {
				System.out.println(jarInfo.toString());
			}

			//POST /jars/<appName>
			client.uploadSparkJobJar(new File("d:\\spark-examples_2.10-1.0.2.jar"), "spark-test");
			
			//GET /contexts
			List<String> contexts = client.getContexts();
			System.out.println("Current contexts:");
			for (String cxt: contexts) {
				System.out.println(cxt);
			}
			
			//POST /contexts/<name>--Create context with name ctxTest and null parameter
			client.createContext("ctxTest", null);
			//POST /contexts/<name>--Create context with parameters
			Map<String, String> params = new HashMap<String, String>();
			params.put(ISparkJobServerClientConstants.PARAM_MEM_PER_NODE, "512m");
			params.put(ISparkJobServerClientConstants.PARAM_NUM_CPU_CORES, "10");
			client.createContext("cxtTest2", params);
			
			//DELETE /contexts/<name>
			client.deleteContext("ctxTest");
			
			//GET /jobs
			List<SparkJobInfo> jobInfos = client.getJobs();
			System.out.println("Current jobs:");
			for (SparkJobInfo jobInfo: jobInfos) {
				System.out.println(jobInfo);
			}
			
			//Post /jobs---Create a new job
			params.put(ISparkJobServerClientConstants.PARAM_APP_NAME, "spark-test");
			params.put(ISparkJobServerClientConstants.PARAM_CLASS_PATH, "spark.jobserver.WordCountExample");
			//1.start a spark job asynchronously and just get the status information
			SparkJobResult result = client.startJob("input.string= fdsafd dfsf blullkfdsoflaw fsdfs", params);
			System.out.println(result);
			
			//2.start a spark job synchronously and wait until the result
			params.put(ISparkJobServerClientConstants.PARAM_CONTEXT, "cxtTest2");
			params.put(ISparkJobServerClientConstants.PARAM_SYNC, "true");
			result = client.startJob("input.string= fdsafd dfsf blullkfdsoflaw fsdffdsfsfs", params);
			System.out.println(result);
			
			//GET /jobs/<jobId>---Gets the result or status of a specific job
			result = client.getJobResult("fdsfsfdfwfef");
			System.out.println(result);
			
			//GET /jobs/<jobId>/config - Gets the job configuration
			SparkJobConfig jobConfig = client.getConfig("fdsfsfdfwfef");
			System.out.println(jobConfig);
		} catch (SparkJobServerClientException e1) {
			e1.printStackTrace();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}