spark汇总

描述

Apache Spark 是用于大规模数据处理的统一分析引擎。它提供 Java、Scala、Python 和 R 中的高级 API，以及支持通用执行图的优化引擎。它还支持一组丰富的高级工具，包括用于 SQL 和结构化数据处理的Spark SQL 、用于机器学习的MLlib、用于图形处理的 GraphX 以及用于增量计算和流处理的结构化流。

1. Spark Core
Spark的核心，是Spark运行的基础。Spark Core以RDD为数据抽象，提供Python、Java、Scala、R语言的API，可以编程进行海量离线数据批处理计算。
2. Spark SQL
Spark SQL是Spark用来操作结构化数据的组件。通过Spark SQL对数据进行处理。
3. Spark Streaming
Spark Streaming是Spark平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的API。
4. Spark MLlib
MLlib是Spark提供的一个机器学习算法库。MLlib不仅提供了模型评估、数据导入等额外的功能，还提供了一些更底层的机器学习原语。
5. Spark GraphX
GraphX是Spark面向图计算提供的框架与算法库。

运行模式

1. Windows模式

多用于本地测试，不需要虚拟机或服务器。

代码示例

WordCount.scala

package com.wunaiieq//1.导入SparkConf,SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]): Unit = {//2.构建SparkConf对象，并设置本地运行和程序的名称val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//3.通过SparkConf对象构建SparkContext对象val sc = new SparkContext(conf)//4.读取文件，并生成RDD对象val fileRdd: RDD[String] = sc.textFile("data/words.txt")//5.将单词进行切割，得到一个存储全部单词的集合对象val wordsRdd: RDD[String] = fileRdd.flatMap(_.split(" "))//6.将单词转换为Tuple2对象（"hello"->("hello",1)）val wordAndOneRdd: RDD[(String, Int)] = wordsRdd.map((_, 1))//7.将元组的value按照key进行分组，并对该组所有的value进行聚合操作val resultRdd: RDD[(String, Int)] = wordAndOneRdd.reduceByKey(_ + _)//8.通过collect方法收集RDD数据val wordCount: Array[(String, Int)] = resultRdd.collect()//9.输出结果wordCount.foreach(println)}
}

log4j.properties
这个没什么说的直接复制用即可

# Set everything to be logged to the console
log4j.rootCategory=ERROR, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{MM/dd HH:mm:ss} %p %c{1}: %m%n# Set the default spark-shell/spark-sql log level to WARN. When running the
# spark-shell/spark-sql, the log level for these classes is used to overwrite
# the root logger's log level, so that the user can have different defaults
# for the shell and regular Spark apps.
log4j.logger.org.apache.spark.repl.Main=WARN
log4j.logger.org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver=WARN# Settings to quiet third party logs that are too verbose
log4j.logger.org.sparkproject.jetty=WARN
log4j.logger.org.sparkproject.jetty.util.component.AbstractLifeCycle=ERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO
log4j.logger.org.apache.parquet=ERROR
log4j.logger.parquet=ERROR# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR# For deploying Spark ThriftServer
# SPARK-34128：Suppress undesirable TTransportException warnings involved in THRIFT-4805
log4j.appender.console.filter.1=org.apache.log4j.varia.StringMatchFilter
log4j.appender.console.filter.1.StringToMatch=Thrift error occurred during processing of message
log4j.appender.console.filter.1.AcceptOnMatch=false

2. Local模式

一台服务器或虚拟机搞定，所谓的Local模式，就是不需要其他任何节点资源就可以在本地执行Spark代码的环境，一般用于教学，调试，演示等。

# 进入spark根目录
cd /opt/module/spark/bin
# 运行视频spark-shell
./spark-shell

webUI

[atguigu@master bin]$ jps
2081 SparkSubmit
2206 Jps
[atguigu@master bin]$ netstat -anp|grep 2081
(Not all processes could be identified, non-owned process infowill not be shown, you would have to be root to see it all.)
tcp6       0      0 192.168.16.100:42050    :::*                    LISTEN      2081/java           
tcp6       0      0 :::4040                 :::*                    LISTEN      2081/java           
tcp6       0      0 192.168.16.100:35770    :::*                    LISTEN      2081/java           
unix  2      [ ]         STREAM     CONNECTED     33071    2081/java            
unix  2      [ ]         STREAM     CONNECTED     36801    2081/java        

浏览器访问

http://192.168.16.100:4040/

spark-submit
以下为使用spark提交jar包示例

./spark-submit --master local[2] --class org.apache.spark.examples.SparkPi /opt/module/spark/examples/jars/spark-examples_2.12-3.1.1.jar 100

参数	描述
--class	要执行程序的主类，可以更换为自己写的应用程序的主类名称
--master local[2]	部署模式，默认为本地模式；数字 2 表示分配的虚拟 CPU 核数量
spark-examples_2.12-3.2.1.jar	运行的应用类所在的 jar 包，实际使用时可以设定为自己打的 jar 包
20	程序的入口参数，根据应用程序的需要，可以是任何有效的输入值

几种提交方式比较

工具	功能	特点	使用场景
bin/spark-submit	提交 Java/Scala/Python/R 代码到 Spark 中运行	提交代码用	正式场合，正式提交 Spark 程序运行
bin/spark-shell	提供一个 Scala 解释器环境，用来以 Scala 代码执行 Spark 程序	解释器环境，写一行执行一行	测试、学习、写一行执行一行、用来验证代码等
bin/pyspark	提供一个 Python 解释器环境，用来以 Python 代码执行 Spark 程序	解释器环境，写一行执行一行	测试、学习、写一行执行一行、用来验证代码等

3. Standalone模式

Standalone是Spark自带的一个资源调度框架，它支持完全分布式，也支持HA

• Master角色：管理整个集群的资源，主要负责资源的调度和分配，并进行集群的监控等职责；并托管运行各个任务的Driver。如Yarn的ResourceManager。
• Worker角色：每个从节点分配资源信息给Worker管理，管理单个服务器的资源类，分配对应的资源来运行Executor(Task)；资源信息包含内存Memory和CPU
  Cores核数。如Yarn的NodeManager。
• Driver角色，管理单个Spark任务在运行的时候的工作，如Yarn的ApplicationMaster “
• Executor角色，单个任务运行的时候的一堆工作者，干活的。它是集群中工作节点（Worker）中的一个JVM进程，负责在 Spark 作业中运行具体任务（Task），任务彼此之间相互独立。Spark 应用启动时，Executor节点被同时启动，并且始终伴随着整个 Spark应用的生命周期而存在。如果有Executor节点发生了故障或崩溃，Spark应用也可以继续执行，会将出错节点上的任务调度到其他Executor节点上继续运行。
  Executor有两个核心功能：
  1.负责运行组成Spark应用的任务，并将结果返回给驱动器进程。
  2.它们通过自身的块管理器（Block Manager）为用户程序中要求缓存的
  RDD 提供内存式存储。RDD 是直接缓存在Executor进程内的，因此任务可以在运行时充分利用缓存数据加速运算。

总结

资源管理维度
集群资源管理者：Master
单机资源管理者：Worker

任务计算维度
单任务管理者：Driver
单任务执行者：Executor

注：Executor运行于Worker进程内,由Worker提供资源供给它们运行

扩展：历史服务器HistoryServer(可选)，Spark Application运行完成以后，保存事件日志数据至HDFS，启动HistoryServer可以查看应用运行相关信息。

4. Yarn模式
Hadoop生态圈里面的一个资源调度框架，Spark也是可以基于Yarn来计算的。
5. 云服务模式（运行在云平台上）
Kubernetes(K8S)容器模式
Spark中的各个角色运行在Kubernetes的容器内部，并组成Spark集群环境。容器化部署是目前业界很流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是(K8S)，而Spark也在新版本中支持了k8s部署模式。
6. Mesos
Mesos是Apache下的开源分布式资源管理框架，它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署，但是在国内，依然使用着传统的Hadoop大数据框架，所以国内使用Mesos框架的并不多。

模式	Spark安装机器数	需启动的进程	所属者	应用场景
Local	1	无	Spark	测试
Standalone	3	Master及Worker	Spark	单独部署
Yarn	1	Yarn及HDFS	Hadoop	混合部署

RDD

描述

Spark RDD（Resilient Distributed Dataset，弹性分布式数据集）代表一个不可变、可分区、元素可并行计算的集合，是Spark进行数据处理的基本单元。

• 不可变性：RDD一旦创建，其数据就不可改变。对RDD的所有操作（如map、filter、reduce等）都会生成一个新的RDD，而不会修改原始RDD。这种不可变性使得RDD在分布式计算环境下非常稳定，避免了并发冲突。
• 可分区性：RDD可以分成多个分区（Partition），每个分区就是一个数据集片段。一个RDD的不同分区可以保存到集群中的不同节点上，从而可以在集群中的不同节点上进行并行计算。分区是Spark作业并行计算的基本单位，每个分区都会被一个计算任务处理，分区的数量决定了并行计算的粒度。
• 弹性：RDD具有弹性容错的特点。当运算中出现异常情况导致分区数据丢失或运算失败时，可以根据RDD的血统（Lineage）关系对数据进行重建。此外，RDD的数据可以保存在内存中，内存放不下时也可以保存在磁盘中，实现了存储的弹性。

特性

1. 分区（Partitions） 含义：RDD的数据被划分为多个分区，每个分区是一个数据块，分布在集群的不同节点上。 作用：每个分区会被一个计算任务处理，分区的数量决定了并行计算的粒度。用户可以在创建RDD时指定分区数，如果没有指定，Spark会根据集群的资源自动设置。
示例：从HDFS文件创建RDD时，默认分区数为文件的Block数。
2. 计算函数（Compute Function） 含义：RDD的计算方法会作用到每个分区上。 作用：当对RDD进行操作（如map、filter等）时，Spark会对每个分区应用这个函数。
示例：在map操作中，计算函数会对每个元素执行指定的转换逻辑。
3. 依赖关系（Dependencies） 含义：RDD之间存在依赖关系。 作用：在部分分区数据丢失时，Spark可以利用依赖关系重新计算丢失的数据，而不是重新计算整个RDD，提高了容错能力。
分类：依赖关系分为窄依赖（Narrow Dependency）和宽依赖（Wide
Dependency）。窄依赖指一个父RDD的分区最多被一个子RDD的分区使用；宽依赖指一个父RDD的分区被多个子RDD的分区使用。
4. 分区器（Partitioner，可选，只有kv型RDD才有） 含义：对于键值对（Key-Value）类型的RDD，可以指定一个分区器来决定数据的分区方式。
作用：分区器决定了数据在集群中的分布，影响并行计算的性能。
类型：Spark支持多种分区器，如HashPartitioner（基于哈希值分区）和RangePartitioner（基于范围分区）。
5. 优先位置（Preferred Locations，可选） 含义：RDD分区规划应当尽量靠近数据所在的服务器 作用：Spark在进行任务调度时，会优先将数据分配到其存储位置进行计算，减少数据传输开销，提高计算效率。
示例：对于HDFS文件，优先位置通常是文件块所在的节点。

RDD创建

1. 通过并行化集合创建，将本地集合对象转分布式RDD

val sc = new SparkContext(conf)
val rdd1:RDD[Int]=sc.parallelize(List(1, 2, 3, 4, 5, 6), 3)
rdd1.glom().collect()

makeRdd()创建，本质上也是使用sc.parallelize(…)

def makeRDD[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism): RDD[T] = withScope {parallelize(seq, numSlices)
}

2. 读取外部数据源 (比如：读取文件 )

//通过SparkConf对象构建SparkContext对象
val sc = new SparkContext(conf)
//读取文件
val fileRdd:RDD[String] = sc.textFile("data/words.txt")

程序执行入口：SparkContext对象
Spark RDD 编程的程序入口对象是SparkContext对象(Scala、Python、Java都是如此)
只有构建出SparkContext, 基于它才能执行后续的API调用和计算
本质上, SparkContext对编程来说, 主要功能就是创建第一个RDD出来。

代码示例（并行化创建）

package com.wunaiieq//1.导入SparkConf类、SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object CreateByParallelize {def main(args: Array[String]): Unit = {//2.构建SparkConf对象。并设置本地运行和程序的名称,*表示使用全部cpu内核,可以指定数量val sparkconf = new SparkConf().setMaster("local[*]").setAppName("CreateRdd1")//3.构建SparkContext对象val sparkContext = new SparkContext(sparkconf)//4.通过并行化创建RDD对象：将本地集合->分布式的RDD对象,如果不指定分区，则根据cpu内核数进行自动分配val rdd: RDD[Int] = sparkContext.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8),3)//5.输出默认的分区数println("默认分区数："+rdd.getNumPartitions)//已经指定为3//6.collect方法：将rdd对象中每个分区的数据，都发送到Driver，形成一个Array对象val array1: Array[Int] = rdd.collect()println("rdd.collect()="+array1.mkString(","))//7.显示出rdd对象中元素被分布到不同分区的数据信息val array2: Array[Array[Int]] = rdd.glom().collect()println("rdd.glom().collect()的内容是:")for(eleArr<- array2){println(eleArr.mkString(","))}}
}

代码示例（读取外部数据）

package com.wunaiieq//1.导入SparkConf,SparkContext类
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object CreateByTextFile {def main(args: Array[String]): Unit = {//2.构建SparkConf对象，并设置本地运行和程序名val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("textFile")//3.通过sparkconf创建SparkContext对象val sparkContext = new SparkContext(sparkConf)//4.通过textFile读取文件//4.1.读取hdfs分布式文件系统上的文件
//    val hdfsRdd: RDD[String] = sparkContext.textFile("hdfs://192.168.16.100:9820/input/data.txt")
//    val hdfsResult: Array[String] = hdfsRdd.collect()
//    println("hdfsRdd分区数"+hdfsRdd.getNumPartitions)
//    println("hdfsRdd内容"+hdfsResult.mkString(","))//4.2读取本地文件val localRdd1: RDD[String] = sparkContext.textFile("data/words.txt")println("localRdd1分区数"+localRdd1.getNumPartitions)println("localRdd1内容"+localRdd1.collect().mkString(","))//5.设置最小分区数val localRdd2: RDD[String] = sparkContext.textFile("data/words.txt",3)println("localRdd2分区数"+localRdd2.getNumPartitions)println("localRdd2内容"+localRdd2.collect().mkString(","))//6.最小分区数设置是一个参考值，Spark会有自己的判断，值太大Spark不会理会val localRdd3: RDD[String] = sparkContext.textFile("data/words.txt", 100)println("localRdd3的分区数"+localRdd3.getNumPartitions)}
}

代码示例（读取目录下的所有文件）

package com.wunaiieq//1.导入类
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object CreateByWholeTextFiles {def main(args: Array[String]): Unit = {//2.构建SparkConf对象，并设置本地运行和程序名称val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WholeTextFiles")//3.使用sparkconf对象构建SparkContet对象val sparkContext = new SparkContext(sparkConf)//5.读取指定目录下的小文件val rdd: RDD[(String, String)] = sparkContext.wholeTextFiles("data")val tuples: Array[(String, String)] = rdd.collect()tuples.foreach(ele=>println(ele._1,ele._2))//6.获取小文件中的内容val array: Array[String] = rdd.map(_._2).collect()println("---------------------------")println(array.mkString("|"))//4.关闭sparkContext对象sparkContext.stop()}
}

算子

详见如下专题RDD算子集合

DAG

详见如下专题DAG专题

SparkSQL

详见如下专题SparkSQL专题

SparkStreaming

详见如下专题SparkStreaming专题