Spark學習筆記1——第一個Spark程序：單詞數統計

Spark學習筆記1——第一個Spark程序：單詞數統計

筆記摘抄自 [美] Holden Karau 等著的《Spark快速大數據分析》

添加依賴

經過 Maven 添加 Spark-core_2.10 的依賴

程序

找了一篇註釋比較清楚的博客代碼¹，一次運行經過

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 目錄下便可運行

---

1. 摘自 https://www.cnblogs.com/itboys/p/6674132.html 【大蔥拌豆腐】↩