Spark學習筆記1——第一個Spark程序：單詞數統計
筆記摘抄自 [美] Holden Karau 等著的《Spark快速大數據分析》
添加依賴
通過 Maven 添加 Spark-core_2.10 的依賴
程序
找了一篇注釋比較清楚的博客代碼[1]，一次運行通過
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 WordCount {
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();
}
}
打包
將程序打包后上傳到Linux
測試用例
[root@server1 ~]# vi test.txt
a
a b
a b c
a b c d
a b c d e
運行
[root@server1 spark-2.4.4-bin-hadoop2.7]# pwd
/root/spark-2.4.4-bin-hadoop2.7
[root@server1 spark-2.4.4-bin-hadoop2.7]# bin/spark-submit --class WordCount ~/SparkTest1.jar ~/test.txt
19/09/09 17:04:56 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
...
19/09/09 17:04:58 INFO DAGScheduler: ResultStage 1 (collect at WordCount.java:97) finished in 0.082 s
19/09/09 17:04:58 INFO DAGScheduler: Job 0 finished: collect at WordCount.java:97, took 0.562887 s
d: 2
e: 1
a: 5
b: 4
c: 3
...
P.s. Spark 單機模式在官網下載壓縮包解壓進入 bin 目錄下即可運行