【大数据】Spark学习笔记

初识Spark

Spark和Hadoop

	Hadoop	Spark
起源时间	2005	2009
起源地	MapReduce	University of California Berkeley
数据处理引擎	Batch	Batch
编程模型	MapReduce	Resilient distributed Datesets
内存管理	Disk Based	JVM Managed
延迟	高	中
吞吐量	中	高
优化机制	手动	手动
API	Low level	high level
流处理	NA	Spark Streaming
SQL支持	Hive, Impala	SparkSQL
Graph支持	NA	GraphX
机器学习支持	NA	SparkML

Spark对比Hadoop特点

Spark优缺点

1. Spark将运算的中间数据存放在内存, 迭代计算效率更高; 而MapReduce的中间结果需要保存到磁盘
2. Spark容错性更高, 通过弹性分布式数据集RDD来实现高容错; 一部分数据丢失或戳错可以通过数据集的计算过程的血缘关系来实现重建; MapReduce发生错误只能重新计算
3. Spark相比于Hadoop提供了transformation和action这两大类的多功能api, 以及流式处理Spark Streaming模块, 图计算GraphX等等; MapReduce只提供了map和reduce两种操作
4. Spark框架和生态更加复杂, 首先有RDD, 血缘lineage, 执行时的有向无环图DAG/stage划分等, 很多时候都需要根据不同场景分别调优以达到性能要求; 而MapReduce框架及应用较为简单, 但运行较为稳定, 更适合长期稳定运行

Hadoop优缺点

优点:

1. 高可靠性: hadoop可以按位存储和处理数据
2. 高扩展性
3. 高效性: Hadoop能够在节点之间动态的移动数据, 并保证各个节点的动态平衡
4. 高容错性: Hadoop能够保存数据的多个副本, 并且能够自动将失败的任务重新分配

缺点:

1. 不适合低延迟的数据访问
2. 无法高效存储大量小文件
3. 不支持多用户写入及任意文件的修改

基本概念

1. Application: 用户编写的Spark应用程序, 包含了driver程序以及集群上运行的程序代码, 物理机器上涉及了driver, master, worker三个节点
2. RDD(Resilient Distributed Dataset): 弹性分布式数据集是Spark中最基本的数据抽象, 代表了一个不可变, 可分区, 可并行计算的集合. RDD具有数据流模型的特点: 自动容错/位置感知性调度/和可伸缩性. RDD允许用户在执行多个查询时显示地将工作集缓存在内存中, 后续的查询能够重用工作机, 这极大地提升了查询速度. RDD包含:
   1. 分片(Partition): 即苏聚集的基本组成功单位, 对于RDD来说, 每个分片都会被一个计算任务吹了, 并决定并行计算的粒度. 用户可以在创建RDD时指定RDD的分片个数, 如果没有指定, 则会采取默认值即分配到的CPUCore个数
   2. 分区计算函数: Spark中RDD的计算是以分片为单位的, 每个RDD都会实现compute函数以达到这个目的. compute函数会对迭代器进行复合, 不需要保存每次计算结果
   3. 重建: 在部分分区数据丢失时, Spark可以通过这个以来关系重新计算丢失的分区数据, 而不是对RDD的所有分区进行计算
   4. Partitioner即RDD的分片函数: 当前Spark中实现了两种类型的分片函数, 一个是基于哈希的HashPartitioner, 另一个是基于范围的RangePartitioner. 只有对于key-value的RDD, 才会有Partitioner
   5. 优先位置(preferedlocation): 对于一个HDFS文件来说, 这个列表保存的就是每个Partition所在的块的位置, 按住奥"移数据不如移动计算"的理念, Spark在记性任务调度的时候, 会尽可能地讲计算任务分配到其所要处理的块的位置
3. DAG: 有向无环图
4. Task: 被发送到executor上的工作单元, 每个Task负责一个分区的数据
5. ShuffleMapTask: 输出是shuffle所需的数据, stage的划分也以此为依据, shuffle之前的所有变换是一个stage, shuffle之后的操作是另个一个stage
6. resultTask: 输出是计算结果
7. Job: 一个Job包含多个RDD及作用于RDD上面的各种操作; 他包含多个task的并行计算, 可以理解为SparkRDD里面的action, 每个action的出发会生成一个job. 用户提交的job会提交给DAGSCheduler; job会被分解为Stage, Stage会被细化乘Task, Task就是每个Partition上的单个数据处理流程
8. Stage: 是job的基本调度单位, 一个Job会分为多组Task, 每组Task被称为一个Stage就行MapStage, ReduceStage,或者也被称为TaskSet, 代表一组关联的, 相互之间没有Shuffle依赖关系的组成的任务集
9. Partition: Partition类似hadoop的Split，计算是以partition为单位进行的
10. Cluster Manager: 指的是在集群上获取资源的外部服务。主要有三种类型:
    1. Standalon : spark原生的资源管理，由Master负责资源的分配。
    2. Apache Mesos:与hadoop MR兼容性良好的一种资源调度框架。
    3. Hadoop Yarn: 主要是指Yarn中的Resource Manager。

Spark 组成

1. Spark Core: Spark核心, 所有核心功能均为Spark提供, Spark Core以RDD为数据抽象, 提供Api, 可以支持海量离线数据批处理计算
2. SparkSQL: 基于Spark Core之上, 提供结构化数据的处理模块, 支持以SQL语言对数据的处理, 本身针对离线的计算场景, 同时基于SparkSQL, Spark提供了StructuredStreaming模块, 可以进行数据的流式计算
3. SparkStream: 以SparkCore为基础, 提供数据的流式计算功能
4. MLLib: 以SparkCore为基础, 进行机器学习计算
5. GraphX: 以SparkCore为基础, 进行图计算, 提供了大量图计算相关的Api

Spark运行模式

1. 本地模式(单机): 本地模式是以一个独立的进程, 通过多个线程来模拟整个Spark运行的环境
2. Standalone模式(集群): Spark中各个角色以独立进程的形式存在, 并组成Spark集群环境
3. Hadoop YARN模式(集群): Spark中的各个角色运行在YARN容器内部, 并组成Spark集群环境
4. Kubernetes模式(容器集群): Spark中的各个角色运行在Kubernetes容器内部, 并组成Spark环境

Spark架构

1. Yarn角色分配:
   1. 以资源管理层面: ResoureManger, ResoureManager
   2. 任务计算层面: ApplicationMaster, Task(容器内计算框架的工作角色)
2. Spark角色分配:
   1. Master: 管理集群的资源
   2. Worker: 集群中任何一个可以运行spark应用代码的节点. Worker是物理节点, 可以在上面启动Executor进程 分配节点资源
   3. Driver: Spark中的Driveer是运行Application的main函数, 并且创建了SparkContext; 创建SparkContext的目的是为了准备Spark应用程序的运行环境. 在Spark中SparkContext负责与Cluster Manager通信, 进行资源申请/任务分配和监控等. 当Excutor部分运行完毕后, Driver同时负责将SparkContext关闭 单个任务的管理
   4. Executor: 在每个Worker上为某应用启动的一个进程, 该进程负责运行Task, 并且负责将数据存在内存或磁盘上, 每个任务都有各独立的Executor. Executor是一个执行Task的容器 单个任务的执行

Standalone架构

Standalone模式Spark自带的一种集群模式, 集群由Master和Spark组成. 除了Master和Worker以外, 还可能由HistoryServer, 该进程会在Spark Application运行完成之后, 保存事件日志到HDFS, 启动HistoryServer可以查看应用相关的信息

基本使用

安装 Spark¹

wget https://dlcdn.apache.org/spark/spark-3.4.1/spark-3.4.1-bin-hadoop3.tgz
tar -xvf spark-3.4.1-bin-hadoop3.tgz
sudo mv spark-3.4.1-bin-hadoop3 /usr/local/spark
vim ~/.bashrc
export SPARK_HOME="/usr/local/spark"
/usr/local/spark/bin/spark-shell

Spark shell - Spark Jobs (passnight.local)包含Spark访问界面

使用Spark实现WordCount

package com.passnight.bigdata.spark;
import lombok.Cleanup;
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;
public class WordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("WordCount")
                .setMaster("local");
        @Cleanup JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD data = context.textFile("hdfs://server.passnight.local/test/word list.txt", 10);
        JavaPairRDD result = data.flatMap(line -> Arrays.stream(line.split(" ")).iterator())
                .mapToPair(word -> new Tuple2<>(word, 1)) // 映射成词频
                .reduceByKey(Integer::sum) // 聚合词频
                // 排序
                .mapToPair(Tuple2::swap)
                .sortByKey(false)
                .mapToPair(Tuple2::swap);
        System.out.println("-".repeat(100));
        System.out.println(result.collect());
        System.out.println("-".repeat(100));
    }
}

输出如下(省略了日志)

[(I,4), (like,2), (passnight,2), (love,2), (hadoop,2)]

RDD

RDD基本概念

1. 分布式计算需要的机制, RDD是提供这些机制的一个抽象
   1. 分区控制
   2. Shuffle控制
   3. 数据存储/序列化/发送
   4. 数据计算
2. RDD定义:
   1. Resilient Distributed Dataset(弹性分布式数据集): 是Spark中最基本的数据抽象, 表示一个不可变/可分区/可并行计算的集合, 三个单词分别有以下含义:
   2. Dataset: 一个数据集合, 用于存放数据
   3. Distributed: RDD中的数据是分布式存储的, 可用于分布式计算
   4. Resilient: RDD中的数据可以存储在内存中或者磁盘中
3. RDD的数据具有以下特性:
   1. 不可变: RDD是不可变集合
   2. 分区性: 数据集合被划分为多个部分, 每个部分被称为分区 对于KV型数据可以有分区器; 且数据读取会尽量靠近数据所在地(移动计算而非数据); 分区是RDD的最小存储单位
   3. 并行性: 计算方法是并行的, 计算方法会作用在每个分区上
   4. 依赖性: RDD之间具有相互依赖的关系 RDD有血缘关系
4. RDD在WordCount中的数据流:

RDD创建

RDD可以通过读取文件或集合创建rdd

package com.passnight.bigdata.spark;
import lombok.Cleanup;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import java.util.Arrays;
public class RDDCreation {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("WordCount")
                .setMaster("local[*]");
        @Cleanup JavaSparkContext context = new JavaSparkContext(conf);
        // 通过并行化的方式创建RDD, 默认分区数为核心数
        JavaRDD rdd = context.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9), 3);
        System.out.println("-".repeat(100));
        System.out.println(rdd.collect());
        System.out.println("-".repeat(100));
        // 也可以通过本地文件创建; 这里的最小分区数是参考值, 而非强制值
        JavaRDD rdd1 = context.textFile("bigdata/src/main/resources/word list.txt", 100);
        System.out.println("-".repeat(100));
        System.out.println(rdd1.getNumPartitions());
        System.out.println("-".repeat(100));
        System.out.println(rdd1.collect());
        System.out.println("-".repeat(100));
        // 从hdfs读取文件
        JavaRDD rdd2 = context.textFile("hdfs://server.passnight.local/test/word list.txt");
        System.out.println("-".repeat(100));
        System.out.println(rdd2.getNumPartitions());
        System.out.println("-".repeat(100));
        System.out.println(rdd2.collect());
        System.out.println("-".repeat(100));
    }
        // 读取多个小文件
        JavaPairRDD rdd3 = context.wholeTextFiles("bigdata/src/main/resources");
        System.out.println("-".repeat(100));
        System.out.println(rdd3.getNumPartitions());
        System.out.println("-".repeat(100));
        System.out.println(rdd3.collect());
        System.out.println("-".repeat(100));
}

输出为:

# 这里省略了日志和分隔符
[1, 2, 3, 4, 5, 6, 7, 8, 9]
61
[I love passnight, I like passnight, I love hadoop, I like hadoop]
2
[I love passnight, I like passnight, I love hadoop, I like hadoop]
[(file:/************/bigdata/src/main/resources/word list.txt,I love passnight
I like passnight #......................

Transformation算子

1. Transformation算子: 返回值仍是一个RDD的算子 这类算子是lazy加载的, 如果没有action算子, 这类算子是不工作的; 如flatMap是一类典型的Transformation算子
2. Action算子: 返回值不是RDD的算子 例如collect

map算子

功能: map算子, 是将RDD中的数字逐条处理, 返回新的RDD

class Map {
    public static void main(String[] args) {
        @Cleanup JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("Map")
                .setMaster("local[*]"));
        List rdd = context.parallelize(IntStream.range(0, 10).boxed().collect(Collectors.toList()), 3)
                .map(i -> i * 10)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
    }
}

输出为:

计算结果:
 [0, 10, 20, 30, 40, 50, 60, 70, 80, 90]

flatMap算子

功能: 先对rdd进行map操作, 再摊平嵌套

class FlatMap {
    public static void main(String[] args) {
        @Cleanup JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("FlatMap")
                .setMaster("local[*]"));
        List rdd = context.parallelize(Arrays.asList("1 2 3", "4 5 6", "7 8 9"), 3)
                .flatMap(line -> Arrays.stream(line.split(" ")).iterator())
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
    }
}

输出为:

计算结果:
 [1, 2, 3, 4, 5, 6, 7, 8, 9]

可以看到多个数组被摊平为一个数组

reduceByKey算子

功能: 针对KV型RDD, 先对key进行分组, 然后根据提供的聚合逻辑, 完成组内数据的聚合操作

class ReduceByKey {
    public static void main(String[] args) {
        @Cleanup
        JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("ReduceByKey")
                .setMaster("local[*]"));
        List> rdd = context.parallelizePairs(Stream.of(1, 1, 1, 2, 2, 2, 3, 4, 4, 3, 10)
                        .map(i -> new Tuple2<>(String.format("值: %d", i), i))
                        .collect(Collectors.toList()), 3)
                .reduceByKey(Integer::sum)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
    }
}

输出结果为:

计算结果:
 [(值: 4,8), (值: 1,3), (值: 2,6), (值: 10,10), (值: 3,6)]

可以看到不同值被分组, 然后进行求和

mapToValues算子

功能: 针对二元元组RDD, 对其内部的Value进行map操作

class MapToValues {
    public static void main(String[] args) {
        @Cleanup
        JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("MapToValues")
                .setMaster("local[*]"));
        List> rdd = context.parallelizePairs(Stream.of(1, 1, 1, 2, 2, 2, 3, 4, 4, 3, 10)
                        .map(i -> new Tuple2<>(String.format("值: %d", i), i))
                        .collect(Collectors.toList()), 3)
                .mapValues(i -> i * 10)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
    }
}

输出结果为:

计算结果:
 [(值: 1,10), (值: 1,10), (值: 1,10), (值: 2,20), (值: 2,20), (值: 2,20), (值: 3,30), (值: 4,40), (值: 4,40), (值: 3,30), (值: 10,100)]

可以看到只有值发生了变化, 且变为了原来的10倍

groupBy算子

功能: 将RDD的数据进行分组

class GroupBy {
    public static void main(String[] args) {
        @Cleanup JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("GroupBy")
                .setMaster("local[*]"));
        List>>> rdd = context.parallelizePairs(Arrays.asList(
                        Tuple2.apply("a", 1), Tuple2.apply("b", 2), Tuple2.apply("b", 1), Tuple2.apply("a", 3), Tuple2.apply("c", 1)
                ), 3)
                .groupBy(Tuple2::_1)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
    }
}

输出结果为:

计算结果:
 [(c,[(c,1)]), (a,[(a,1), (a,3)]), (b,[(b,2), (b,1)])]

可以看到已经根据key分组了

ffilter算子

功能: 过滤符合条件的数据

class Filter {
    public static void main(String[] args) {
        @Cleanup JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("Filter")
                .setMaster("local[*]"));
        List rdd = context.parallelize(IntStream.range(0, 10).boxed().collect(Collectors.toList()), 3)
                .filter(i -> i % 2 == 0)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
    }
}

输出结果为:

计算结果:
 [0, 2, 4, 6, 8]

可以看到已将偶数都过滤出来了

distinct算子

功能: 将rdd数据去重

class Distinct {
    public static void main(String[] args) {
        @Cleanup JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("Distinct")
                .setMaster("local[*]"));
        List rdd = context.parallelize(Arrays.asList(1, 1, 1, 2, 2, 2, 3, 3, 3), 3)
                .distinct(2)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd);
        
        List> rdd2 = context.parallelizePairs(Arrays.asList(Tuple2.apply("a", 1),
                        Tuple2.apply("b", 1), Tuple2.apply("b", 1),
                        Tuple2.apply("a", 3), Tuple2.apply("a", 1)), 3)
                .distinct(2)
                .collect();
        System.out.printf("计算结果:%n %s%n", rdd2);
    }
}

输出结果为:

计算结果:
 [2, 1, 3]
计算结果:
 [(a,1), (a,3), (b,1)]

可以看到无论是KV型数据还是普通的数据, 都已经去重了

union算子

功能: 将两个rdd合并成一个rdd

class Union {
    public static void main(String[] args) {
        @Cleanup JavaSparkContext context = new JavaSparkContext(new SparkConf()
                .setAppName("Union")
                .setMaster("local[*]"));
        JavaRDD