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

時間  2019-11-17
標籤 spark 學習 筆記 rdd 欄目 Spark 简体版
原文   原文鏈接

目錄java

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

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

向Spark傳遞函數

大部分 Spark 的轉化操做和一部分行動操做，都須要傳遞函數後進行計算。如何傳遞函數下文將用 Java 展現。shell

Java 向 Spark 傳遞函數須要實現 Spark 的 org.apache.spark.api.java.function 包中的接口。一些基本的接口以下表：apache

函數名 實現的方法 用途
Function<T, R> R call(T) 接收一個輸入值並返回一個輸出值，用於相似map() 和
filter() 等操做中
Function2<T1, T2, R> R call(T1, T2) 接收兩個輸入值並返回一個輸出值，用於相似aggregate()
和fold() 等操做中
FlatMapFunction<T, R> Iterable call(T) 接收一個輸入值並返回任意個輸出，用於相似flatMap()
這樣的操做中

經過匿名內部類

見上篇筆記例程。api

經過具名類傳遞

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 支持聚合數據的鍵值對操做等等。oop

基本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中flatmap和map的區別.jpg

僞集合操做

RDD 不算是嚴格意義上的集合，可是一些相似集合的屬性讓它可以支持許多集合操做，下圖展現了常見的集合操做：

簡單的僞集合操做.jpg

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

RDD間笛卡爾積的操做.jpg

如下對基本 RDD 的轉化操做進行梳理：

  • 單個 RDD {1，2，3，3} 的轉化操做
函數名 目的 示例 結果
map() 將函數應用於RDD 中的每一個元
素，將返回值構成新的RDD		 rdd.map(x => x + 1) {2, 3, 4, 4}
flatMap() 將函數應用於RDD 中的每一個元
素，將返回的迭代器的全部內
容構成新的RDD。一般用來切
分單詞		 rdd.flatMap(x => x.to(3)) {1, 2, 3, 2, 3, 3, 3}
filter() 返回一個由經過傳給filter()
的函數的元素組成的RDD		 rdd.filter(x => x != 1) {2, 3, 3}
distinct() 去重 rdd.distinct() {1, 2, 3}
sample(withReplacement, fraction, [seed]) 對RDD 採樣，以及是否替換 rdd.sample(false, 0.5) 非肯定的
  • 兩個 RDD {1,2,3},{3,4,5}的 RDD 的轉化操做
函數名 目的 示例 結果
union() 生成一個包含兩個RDD 中全部元
素的RDD		 rdd.union(other) {1, 2, 3, 3, 4, 5}
intersection() 求兩個RDD 共同的元素的RDD rdd.intersection(other) {3}
subtract() 移除一個RDD 中的內容（例如移
除訓練數據）		 rdd.subtract(other) {1, 2}
cartesian() 與另外一個RDD 的笛卡兒積 rdd.cartesian(other) {(1, 3), (1, 4), ...
(3, 5)}

行動操做

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 = Arrays.asList(1, 2, 3, 4, 5);
        JavaRDD<Integer> originRDD = javaSparkContext.parallelize(data);

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

        //reduceByKey，按照相同的key進行reduce操做
        List<String> list = Arrays.asList("key1", "key1", "key2", "key2", "key3");
        JavaRDD<String> stringRDD = javaSparkContext.parallelize(list);
        //轉爲key-value形式
        JavaPairRDD<String, Integer> 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 = Arrays.asList(1, 2, 3, 4, 5);
        JavaRDD<Integer> rdd = javaSparkContext.parallelize(data);
        AvgCount initial = new AvgCount(0, 0);
        Function2<AvgCount, Integer, AvgCount> addAndCount =
                new Function2<AvgCount, Integer, AvgCount>() {
                    public AvgCount call(AvgCount a, Integer x) {
                        a.total += x;
                        a.num += 1;
                        return a;
                    }
                };
        Function2<AvgCount, AvgCount, AvgCount> combine =
                new Function2<AvgCount, AvgCount, AvgCount>() {
                    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
...

經常使用的行動操做整理

函數名 目的 示例 結果
collect() 返回RDD 中的全部元素 rdd.collect() {1, 2, 3, 3}
count() RDD 中的元素個數 rdd.count() 4
countByValue() 各元素在RDD 中出現的次數 rdd.countByValue() {(1, 1),
(2, 1),
(3, 2)}
take(num) 從RDD 中返回num 個元素 rdd.take(2) {1, 2}
top(num) 從RDD 中返回最前面的num
個元素		 rdd.top(2) {3, 3}
takeOrdered(num)
(ordering)		 從RDD 中按照提供的順序返
回最前面的num 個元素		 rdd.takeOrdered(2)(myOrdering) {3, 3}
takeSample(withReplace
ment, num, [seed])		 從RDD 中返回任意一些元素 rdd.takeSample(false, 1) 非肯定的
reduce(func) 並行整合RDD 中全部數據
（例如sum）		 rdd.reduce((x, y) => x + y) 9
fold(zero)(func) 和reduce() 同樣， 可是須要
提供初始值		 rdd.fold(0)((x, y) => x + y) 9
aggregate(zeroValue)
(seqOp, combOp)		 和reduce() 類似， 可是一般
返回不一樣類型的函數		 rdd.aggregate((0, 0))
((x, y) =>
(x._1 + y, x._2 + 1),
(x, y) =>
(x._1 + y._1, x._2 + y._2))		 (9,4)
foreach(func) 對RDD 中的每一個元素使用給
定的函數		 rdd.foreach(func) 無

不一樣 RDD 的類型轉換

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

下表爲 Java 中針對專門類型的函數接口：

函數名 等價函數 用途
DoubleFlatMapFunction Function<T, Iterable > 用於flatMapToDouble，以
生成DoubleRDD
DoubleFunction Function<T, Double> 用於mapToDouble，以生成
DoubleRDD
PairFlatMapFunction<T, K, V> Function<T, Iterable<Tuple2<K, V>>> 用於flatMapToPair，以生
成PairRDD<K, V>
PairFunction<T, K, V> Function<T, Tuple2<K, V>> 用於mapToPair， 以生成
PairRDD<K, V>

例程

以 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 從緩存中移除

持久化級別

級　　別 使用的
空間		 CPU
時間		 是否在
內存中		 是否在
磁盤上		 備註
MEMORY_ONLY 高 低 是 否
MEMORY_ONLY_SER 低 高 是 否
MEMORY_AND_DISK 高 中等 部分 部分 若是數據在內存中放不下，則溢寫到磁盤上
MEMORY_AND_DISK_SER 低 高 部分 部分 若是數據在內存中放不下，則溢寫到磁盤上。在內存中存放序列化後的數據
DISK_ONLY 低 高 否 是

P.s.

能夠經過在存儲級別的末尾加上「_2」來把持久化數據存爲兩份

  1. 摘自天涯淚小武 的博客↩

  2. fold() 和 reduce() 不一樣的是，須要再加上一個「初始值」來做爲每一個分區第一次調用時的結果；aggregate() 和 前二者不一樣的是，返回值類型能夠和 RDD 的類型不一致↩

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程