Spark运行任务

Java

import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.clustering.KMeans;
import org.apache.spark.mllib.clustering.KMeansModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.SparkConf;/* Test:* sh start-dfs.sh* sh start-yarn.sh* cd /appl/spark-1.4.0/* sbin/start-all.sh* hadoop fs -put /mk/test/kmeans_data.txt /test/* ./bin/spark-submit /mk/test/KMeansSim.jar*/
public class KMeansSim {public static void main(String[] args) {// environment initializationSparkConf conf &＃61; new SparkConf().setAppName("K-means Example");JavaSparkContext sc &＃61; new JavaSparkContext(conf);// Load and parse data (${SPARK_HOME}/data/mllib/kmeans_data.txt)String path &＃61; "/test/kmeans_data.txt";JavaRDD data &＃61; sc.textFile(path);JavaRDD parsedData &＃61; data.map(new Function() {public Vector call(String s) {return Vectors.dense(toDoubleArray(s));}});parsedData.cache();// Cluster the data into two classes using KMeansint numClusters &＃61; 2;int numIterations &＃61; 20;KMeansModel clusters &＃61; KMeans.train(parsedData.rdd(), numClusters, numIterations);// Evaluate clustering by computing Within Set Sum of Squared Errorsdouble WSSSE &＃61; clusters.computeCost(parsedData.rdd());System.out.println("Within Set Sum of Squared Errors &＃61; " &＃43; WSSSE);// Save and load modelclusters.save(sc.sc(), "myModelPath");KMeansModel sameModel &＃61; KMeansModel.load(sc.sc(), "myModelPath");// predict testSystem.out.println("~~~predict:" &＃43; clusters.predict(Vectors.dense(toDoubleArray("1.0 2.1 3.8"))));// endingsc.stop();}// String to double[]public static double[] toDoubleArray(String s) {String[] sarray &＃61; s.split(" ");double[] values &＃61; new double[sarray.length];for (int i &＃61; 0; i }


5. 运行

./bin/spark-submit /mk/test/KMeansSim.jar