Spark學習筆記3——RDD（下）

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

目錄

• Spark學習筆記3——RDD（下）
  • 向Spark傳遞函式
    • 通過匿名內部類
    • 通過具名類傳遞
    • 通過帶引數的 Java 函式類傳遞
      • 例程
      • 測驗文本 test.txt
      • 測驗結果
    • 通過 lambda 運算式傳遞（僅限于 Java 8 及以上）
      • 例程
      • 測驗文本
      • 運行結果
  • 常見的轉化操作和行動操作
    • 基本RDD
      • 針對各個元素的轉化操作
      • 偽集合操作
    • 行動操作
      • reduce() 與 reduceByKey()
      • aggregate()[^2]
      • 常用的行動操作整理
    • 不同 RDD 的型別轉換
  • 持久化

向Spark傳遞函式

大部分 Spark 的轉化操作和一部分行動操作，都需要傳遞函式后進行計算，如何傳遞函式下文將用 Java 展示，

Java 向 Spark 傳遞函式需要實作 Spark 的 org.apache.spark.api.java.function 包中的介面，一些基本的介面如下表：

通過匿名內部類

見上篇筆記例程，

通過具名類傳遞

class ContainsError implements Function<String, Boolean>() {
public Boolean call(String x) { return x.contains("error"); }
}
...
RDD<String> errors = lines.filter(new ContainsError());

• 使用具名類在程式組織比較龐大是顯得比較清晰
• 可以使用建構式如“通過帶引數的 Java 函式類傳遞”中所示

通過帶引數的 Java 函式類傳遞

例程

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;

import java.util.List;

public class Contains implements Function<String, Boolean> {
    private String query;

    public Contains(String query) {
        this.query = query;
    }

    public Boolean call(String x) {
        return x.contains(query);
    }

    public static void main(String[] args) {
        SparkConf sc = new SparkConf().setAppName("Contains");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sc);
        JavaRDD<String> log = javaSparkContext.textFile(args[0]);
        
        JavaRDD<String> content = log.filter(new Contains(args[1]));
        
        List<String> contentList = content.collect();
        for (String output : contentList) {
            System.out.println(output);
        }
        javaSparkContext.stop();
    }
}

測驗文本 test.txt

this is a test
this is a simple test
this is a simple test about RDD
let us check it out

測驗結果

[root@server1 spark-2.4.4-bin-hadoop2.7]# bin/spark-submit --class Contains ~/RDDFuncNamedClass.jar ~/test.txt RDD
...
19/09/16 15:06:50 INFO DAGScheduler: Job 0 finished: collect at Contains.java:24, took 0.445049 s
this is a simple test about RDD
...

通過 lambda 運算式傳遞（僅限于 Java 8 及以上）

例程

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import java.util.List;

public class LambdaTest {
    public static void main(final String[] args) {
        SparkConf sc = new SparkConf().setAppName("Contains");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sc);
        JavaRDD<String> log = javaSparkContext.textFile(args[0]);

        JavaRDD<String> content = log.filter(s -> s.contains(args[1]));

        List<String> contentList = content.collect();
        for (String output : contentList) {
            System.out.println(output);
        }
        javaSparkContext.stop();
    }
}

測驗文本

使用上文同一個文本

運行結果

[root@server1 spark-2.4.4-bin-hadoop2.7]# bin/spark-submit --class Contains ~/RDDFuncNamedClass.jar ~/test.txt check
...
19/09/16 15:27:10 INFO DAGScheduler: Job 0 finished: collect at Contains.java:24, took 0.440515 s
let us check it out
...

常見的轉化操作和行動操作

Spark 中有不同型別的 RDD，不同的 RDD 可以支持不同的操作，

除了基本的RDD外，還有數字型別的 RDD 支持統計型函式操作、鍵值對形式的 RDD 支持聚合資料的鍵值對操作等等，

基本RDD

針對各個元素的轉化操作

為了方便，代碼在 pyspark 中展示：

# map()
# map() 的回傳值型別不需要和輸入型別一樣
>>> nums = sc.parallelize([1, 2, 3, 4])
>>> squared = nums.map(lambda x: x * x).collect()
>>> for num in squared:
...     print "%i " % (num)
... 
1 
4 
9 
16

# flatMap()
# 給flatMap() 的函式被分別應用到了輸入RDD 的每個元素上，
# 回傳的是一個回傳值序列的迭代器，
# 
>>> lines = sc.parallelize(["hello world", "hi"])
>>> words = lines.flatMap(lambda line: line.split(" "))
>>> words.first()
'hello'

map() 和 flatmap() 區別如下：

偽集合操作

RDD 不算是嚴格意義上的集合，但是一些類似集合的屬性讓它能夠支持許多集合操作，下圖展示了常見的集合操作：

此外，RDD 還支持笛卡爾積的操作：

以下對基本 RDD 的轉化操作進行梳理：

• 單個 RDD {1，2，3，3} 的轉化操作

• 兩個 RDD {1,2,3},{3,4,5}的 RDD 的轉化操作

行動操作

reduce() 與 reduceByKey()

例程^[1]

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 scala.Tuple2;

import java.util.Arrays;
import java.util.List;

public class SimpleReduce {
    public static void main(String[] args) {
        SparkConf sc = new SparkConf().setAppName("Contains");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sc);
        List<Integer> data = https://www.cnblogs.com/aar-lee/p/Arrays.asList(1, 2, 3, 4, 5);
        JavaRDD originRDD = javaSparkContext.parallelize(data);

        Integer sum = originRDD.reduce((a, b) -> a + b);
        System.out.println(sum);

        //reduceByKey，按照相同的key進行reduce操作
        List list = Arrays.asList("key1", "key1", "key2", "key2", "key3");
        JavaRDD stringRDD = javaSparkContext.parallelize(list);
        //轉為key-value形式
        JavaPairRDD pairRDD = stringRDD.mapToPair(k -> new Tuple2<>(k, 1));
        List list1 = pairRDD.reduceByKey((x, y) -> x + y).collect();
        System.out.println(list1);
    }

}

運行結果

...
19/09/17 17:08:37 INFO DAGScheduler: Job 0 finished: reduce at SimpleReduce.java:21, took 0.480038 s
15
...
19/09/17 17:08:38 INFO DAGScheduler: Job 1 finished: collect at SimpleReduce.java:29, took 0.237601 s
[(key3,1), (key1,2), (key2,2)]
...

aggregate()^[2]

例程

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;

public class AvgCount implements Serializable {
    private AvgCount(int total, int num) {
        this.total = total;
        this.num = num;
    }
    private int total;
    private int num;
    private double avg() {
        return total / (double) num;
    }

    public static void main(String[] args) {
        SparkConf sc = new SparkConf().setAppName("Contains");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sc);
        List<Integer> data = https://www.cnblogs.com/aar-lee/p/Arrays.asList(1, 2, 3, 4, 5);
        JavaRDD rdd = javaSparkContext.parallelize(data);
        AvgCount initial = new AvgCount(0, 0);
        Function2 addAndCount =
                new Function2() {
                    public AvgCount call(AvgCount a, Integer x) {
                        a.total += x;
                        a.num += 1;
                        return a;
                    }
                };
        Function2 combine =
                new Function2() {
                    public AvgCount call(AvgCount a, AvgCount b) {
                        a.total += b.total;
                        a.num += b.num;
                        return a;
                    }
                };
        AvgCount result = rdd.aggregate(initial, addAndCount, combine);
        System.out.println(result.avg());
    }
}

運行結果

[root@server1 spark-2.4.4-bin-hadoop2.7]# bin/spark-submit --class AvgCount ~/Spark_RDD_Aggregate.jar
...
19/09/18 15:28:19 INFO DAGScheduler: Job 0 finished: aggregate at AvgCount.java:43, took 0.517385 s
3.0
...

常用的行動操作整理

不同 RDD 的型別轉換

Spark 中有些函式只能作用于特定型別的 RDD，例如 mean() 和 variance() 只能處理數值 RDD，join() 只能用于處理鍵值對 RDD，在 Scala 和 Java 中都沒有與之對應的標準 RDD 類，故使用這些函式時必須要確保獲得了正確的專用 RDD 類，（Scala 為隱式轉換）

下表為 Java 中針對專門型別的函式介面：

例程

以 DoubleFunction 為例：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaDoubleRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.DoubleFunction;

import java.util.Arrays;

public class DoubleRDD {
    public static void main(String[] args) {
        SparkConf sparkConf=new SparkConf().setAppName("DoubleRDD");
        JavaSparkContext javaSparkContext=new JavaSparkContext(sparkConf);
        JavaRDD<Integer> rdd = javaSparkContext.parallelize(Arrays.asList(1, 2, 3, 4));
        JavaDoubleRDD result = rdd.mapToDouble(
                new DoubleFunction<Integer>() {
                    public double call(Integer x) {
                        return (double) x * x;
                    }
                });
        System.out.println(result.mean());
    }
}

運行結果

[root@server1 spark-2.4.4-bin-hadoop2.7]# bin/spark-submit --class DoubleRDD ~/Spark_RDD_DoubleRDD.jar
...
19/09/18 16:09:38 INFO DAGScheduler: Job 0 finished: mean at DoubleRDD.java:20, took 0.500705 s
7.5
...

持久化

為了避免多次計算同一個 RDD，我們常常對資料進行持久化處理，具體操作可以參見上一節例程，

Tips：

• 在Scala 和Java 中，默認情況下 persist() 會把資料以序列化的形式快取在JVM 的堆空間中
• 在Python 中，我們會始終序列化要持久化存盤的資料，所以持久化級別默認值就是以序列化后的物件存盤在JVM 堆空間中
• 當我們把資料寫到磁盤或者堆外存盤上時，也總是使用序列化后的資料
• 快取的資料太多，記憶體中放不下，Spark 會自動利用最近最少使用（LRU）的快取策略把最老的磁區從記憶體中移除
• unpersist() 可以手動把持久化的RDD 從快取中移除

持久化級別

P.s.

可以通過在存盤級別的末尾加上“_2”來把持久化資料存為兩份

標籤：大數據

上一篇：Flink入門寶典（詳細截圖版）

下一篇：利用 pandas庫讀取excel表格資料