Spark的Java API例子詳解

package com.hand.study;
 import scala.Tuple2; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.regex.Pattern; public final class JavaWordCount { private static final Pattern SPACE = Pattern.compile(" "); public static void main(String[] args) throws Exception { if (args.length < 1) { System.err.println("Usage: JavaWordCount <file>"); System.exit(1); } /** * 對於全部的spark程序所言，要進行全部的操做，首先要建立一個spark上下文。 * 在建立上下文的過程當中，程序會向集羣申請資源及構建相應的運行環境。 * 設置spark應用程序名稱 * 建立的 sarpkContext 惟一須要的參數就是 sparkConf，它是一組 K-V 屬性對。 */ SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount"); JavaSparkContext ctx = new JavaSparkContext(sparkConf); /** * 利用textFile接口從文件系統中讀入指定的文件，返回一個RDD實例對象。 * RDD的初始建立都是由SparkContext來負責的，將內存中的集合或者外部文件系統做爲輸入源。 * RDD：彈性分佈式數據集，即一個 RDD 表明一個被分區的只讀數據集。一個 RDD 的生成只有兩種途徑， * 一是來自於內存集合和外部存儲系統，另外一種是經過轉換操做來自於其餘 RDD，好比 Map、Filter、Join，等等。 * textFile()方法可將本地文件或HDFS文件轉換成RDD，讀取本地文件須要各節點上都存在，或者經過網絡共享該文件 *讀取一行 */ JavaRDD<String> lines = ctx.textFile(args[0], 1); /** * * new FlatMapFunction<String, String>兩個string分別表明輸入和輸出類型 * Override的call方法須要本身實現一個轉換的方法，並返回一個Iterable的結構 * * flatmap屬於一類很是經常使用的spark函數，簡單的說做用就是將一條rdd數據使用你定義的函數給分解成多條rdd數據 * 例如，當前狀態下，lines這個rdd類型的變量中，每一條數據都是一行String，咱們如今想把他拆分紅1個個的詞的話， * 能夠這樣寫 ： */
        //flatMap與map的區別是，對每一個輸入，flatMap會生成一個或多個的輸出，而map只是生成單一的輸出 //用空格分割各個單詞,輸入一行,輸出多個對象,因此用flatMap
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) { return Arrays.asList(SPACE.split(s)).iterator(); } }); /** * map 鍵值對 ，相似於MR的map方法 * pairFunction<T,K,V>: T:輸入類型；K,V：輸出鍵值對 * 表示輸入類型爲T,生成的key-value對中的key類型爲k,value類型爲v,對本例,T=String, K=String, V=Integer(計數) * 須要重寫call方法實現轉換 */ JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() { //scala.Tuple2<K,V> call(T t) //Tuple2爲scala中的一個對象,call方法的輸入參數爲T,即輸入一個單詞s,新的Tuple2對象的key爲這個單詞,計數爲1
 @Override public Tuple2<String, Integer> call(String s) { return new Tuple2<String, Integer>(s, 1); } }); //A two-argument function that takes arguments // of type T1 and T2 and returns an R.
        /** * 調用reduceByKey方法,按key值進行reduce * reduceByKey方法，相似於MR的reduce * 要求被操做的數據（即下面實例中的ones）是KV鍵值對形式，該方法會按照key相同的進行聚合，在兩兩運算 * 若ones有<"one", 1>, <"one", 1>,會根據"one"將相同的pair單詞個數進行統計,輸入爲Integer,輸出也爲Integer *輸出<"one", 2> */ JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() { //reduce階段，key相同的value怎麼處理的問題
 @Override public Integer call(Integer i1, Integer i2) { return i1 + i2; } }); //備註：spark也有reduce方法，輸入數據是RDD類型就能夠，不須要鍵值對， // reduce方法會對輸入進來的全部數據進行兩兩運算

        /** * collect方法用於將spark的RDD類型轉化爲咱們熟知的java常見類型 */ List<Tuple2<String, Integer>> output = counts.collect(); for (Tuple2<?,?> tuple : output) { System.out.println(tuple._1() + ": " + tuple._2()); } ctx.stop(); } }