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教程