Apache Spark - 用于大规模数据分析的统一引擎

下载
运行示例和 Shell
- 使用 Spark Connect 在 Anywhere 上运行 Spark 客户端应用程序
在集群上启动
从这里去哪里
使用 Spark Shell 进行交互式分析
- 基本
- 有关数据集作的更多信息
- 缓存
自包含应用程序
从这里去哪里
idea中使用
- 可能会遇到下边问题
- - 问题1
  - 问题2
  - 问题3
  - 问题4

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

下载

从项目网站的下载页面获取 Spark。本文档适用于 Spark 版本 3.5.5。Spark 将 Hadoop 的客户端库用于 HDFS 和 YARN。下载内容已针对少数流行的 Hadoop 版本进行了预打包。用户还可以下载“Hadoop 免费”二进制文件，并通过扩充 Spark 的类路径，使用任何 Hadoop 版本运行 Spark。 Scala 和 Java 用户可以使用其 Maven 坐标将 Spark 包含在其项目中，Python 用户可以从 PyPI 安装 Spark。

如果您想从 source，请访问 Building Spark。

Spark 可以在 Windows 和类 UNIX 系统（例如 Linux、Mac OS）上运行，并且它应该可以在运行受支持的 Java 版本的任何平台上运行。这应包括 x86_64 和 ARM64 上的 JVM。在一台机器上本地运行很容易 - 您只需在 system 上安装，或者使用指向 Java 安装的环境变量。java``PATH``JAVA_HOME

Spark 可在 Java 8/11/17、Scala 2.12/2.13、Python 3.8+ 和 R 3.5+ 上运行。版本 8u371 之前的 Java 8 支持从 Spark 3.5.0 开始弃用。使用 Scala API 时，应用程序必须使用编译 Spark 的相同 Scala 版本。例如，使用 Scala 2.13 时，使用为 2.13 编译的 Spark，并编译为 Scala 2.13 的代码/应用程序。

对于 Java 11，需要对 Apache Arrow 库进行设置。这可以防止 Apache Arrow 在内部使用 Netty 时出现错误。-Dio.netty.tryReflectionSetAccessible=true``java.lang.UnsupportedOperationException: sun.misc.Unsafe or java.nio.DirectByteBuffer.(long, int) not available

运行示例和 Shell

Spark 附带了几个示例程序。Python、Scala、Java 和 R 示例位于目录中。examples/src/main

要在 Python 解释器中以交互方式运行 Spark，请使用：bin/pyspark

./bin/pyspark --master "local[2]"

以 Python 提供示例应用程序。例如：

./bin/spark-submit examples/src/main/python/pi.py 10

要运行 Scala 或 Java 示例程序之一，请在顶级 Spark 目录中使用。（在幕后，这个调用更通用的 spark-submit 脚本启动应用程序）。例如bin/run-example <class> [params]

./bin/run-example SparkPi 10

您还可以通过 Scala shell 的修改版本以交互方式运行 Spark。这是一个学习框架的好方法。

./bin/spark-shell --master "local[2]"

该选项指定分布式集群的主 URL，或运行本地使用 1 个线程，或者使用 N 个线程在本地运行。您应该从 using for testing 开始。有关选项的完整列表，请使用选项运行 Spark shell。--master``local``local[N]``local``--help

从版本 1.4 开始，Spark 提供了一个 R API（仅包含数据帧 API）。要在 R 解释器中以交互方式运行 Spark，请使用：bin/sparkR

./bin/sparkR --master "local[2]"

R 中还提供了示例应用程序。例如：

./bin/spark-submit examples/src/main/r/dataframe.R

使用 Spark Connect 在 Anywhere 上运行 Spark 客户端应用程序

Spark Connect 是 Spark 3.4 中引入的一种新的客户端-服务器体系结构，用于分离 Spark 客户端应用程序，并允许远程连接到 Spark 集群。两者之间的分离客户端和服务器允许从任何地方利用 Spark 及其开放式生态系统，嵌入式在任何应用程序中。在 Spark 3.4 中，Spark Connect 为 PySpark 和 Scala 中的数据帧/数据集 API 支持。

要了解有关 Spark Connect 及其使用方法的更多信息，请参阅 Spark Connect 概述。

在集群上启动

Spark 集群模式概述介绍了在集群上运行的关键概念。 Spark 既可以单独运行，也可以在多个现有集群管理器上运行。它目前提供了几个部署选项：

独立部署模式：在私有集群上部署 Spark 的最简单方法
Apache Mesos （已弃用）
Hadoop YARN
Kubernetes （简体中文）

从这里去哪里

编程指南：

快速入门：Spark API 快速介绍;从这里开始！
RDD 编程指南：Spark 基础知识概述 - RDD （核心但旧 API）、累加器和广播变量
Spark SQL、数据集和数据帧：使用关系查询处理结构化数据（比 RDD 更新的 API）
结构化流式处理：使用关系查询处理结构化数据流（使用 Datasets 和 DataFrames，比 DStreams 更新的 API）
Spark Streaming：使用 DStreams（旧 API）处理数据流
MLlib：应用机器学习算法
GraphX：处理图形
SparkR：在 R 中使用 Spark 处理数据
PySpark：在 Python 中使用 Spark 处理数据
Spark SQL CLI：在命令行上使用 SQL 处理数据

API 文档：

Spark Scala API （Scaladoc）
Spark Java API （Javadoc）
Spark Python API （Sphinx）
Spark R API （Roxygen2）
Spark SQL，内置函数（MkDocs）

部署指南：

Cluster Overview：介绍在集群上运行时的概念和构成
提交应用程序：打包和部署应用程序
部署模式：
- Amazon EC2：可让您在大约 5 分钟内在 EC2 上启动集群的脚本
- 独立部署模式：无需第三方集群管理器即可快速启动独立集群
- Mesos：使用 Apache Mesos 部署私有集群
- YARN：在 Hadoop NextGen （YARN）上部署 Spark
- Kubernetes：在 Kubernetes 上部署 Spark

其他文件：

配置：通过其配置系统自定义 Spark
监控：跟踪应用程序的行为
调优指南：优化性能和内存使用的最佳实践
作业调度：在 Spark 应用程序之间和内部调度资源
安全性：Spark 安全支持
硬件配置：集群硬件的建议
与其他存储系统集成：
- 云基础设施
- OpenStack Swift
迁移指南：Spark组件迁移指南
构建 Spark：使用 Maven 系统构建 Spark
为 Spark 做贡献
第三方项目：相关的第三方 Spark 项目

外部资源：

Spark 主页
Spark 社区资源，包括本地聚会
StackOverflow 标签 apache-spark
邮件列表：在此处询问有关 Spark 的问题
AMP Camps：加州大学伯克利分校的一系列训练营，包括讲座和关于 Spark、Spark Streaming、Mesos 等的练习。[视频、](https://www.youtube.com/user/BerkeleyAMPLab/search?query=amp camp) 可免费在线获取。
代码示例：Spark 的子文件夹（Scala、Java、Python、Rexamples)

快速入门 - Spark 3.5.5 文档

Apache spark 下载镜像

使用 Spark Shell 进行交互式分析

基本

./bin/spark-shell
scala> val textFile = spark.read.textFile("README.md")
textFile: org.apache.spark.sql.Dataset[String] = [value: string]

scala> textFile.count() // Number of items in this Dataset
res0: Long = 126 // May be different from yours as README.md will change over time, similar to other outputsscala> textFile.first() // First item in this Dataset
res1: String = # Apache Spark

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: org.apache.spark.sql.Dataset[String] = [value: string]

scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15

有关数据集作的更多信息

scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Int = 15

scala> import java.lang.Math
import java.lang.Mathscala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))
res5: Int = 15

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()
wordCounts: org.apache.spark.sql.Dataset[(String, Long)] = [value: string, count(1): bigint]

scala> wordCounts.collect()
res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)

缓存

scala> linesWithSpark.cache()
res7: linesWithSpark.type = [value: string]scala> linesWithSpark.count()
res8: Long = 15scala> linesWithSpark.count()
res9: Long = 15

访问：Spark shell - Spark Jobs
在这里插入图片描述

自包含应用程序

此示例将使用 Maven 编译应用程序 JAR，但任何类似的构建系统都可以使用。

我们将创建一个非常简单的 Spark 应用程序：SimpleApp.java

/*** @author heliming* @version 1.0* @date 2025/3/24-22:18* @description TODO*/
/* SimpleApp.java */
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;public class SimpleApp {public static void main(String[] args) {System.out.println(123);String logFile = "D:\\spark-3.4.4-bin-hadoop3\\README.md"; // Should be some file on your systemSparkSession spark = SparkSession.builder().appName("Simple Application").getOrCreate();Dataset<String> logData = spark.read().textFile(logFile).cache();long numAs = logData.filter((FilterFunction<String>) s -> s.contains("a")).count();long numBs = logData.filter((FilterFunction<String>) s -> s.contains("b")).count();System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);spark.stop();}
}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project><groupId>edu.berkeley</groupId><artifactId>simple-project</artifactId><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>8</source><target>8</target></configuration></plugin></plugins></build><modelVersion>4.0.0</modelVersion><name>Simple Project</name><packaging>jar</packaging><version>1.0</version><dependencies><dependency> <!-- Spark dependency --><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>3.5.5</version><scope>provided</scope></dependency></dependencies>
</project>

我们根据规范的 Maven 目录结构对这些文件进行布局：

$ find .
./pom.xml
./src
./src/main
./src/main/java
./src/main/java/SimpleApp.java

现在，我们可以使用 Maven 打包应用程序并使用 ../bin/spark-submit

# Package a JAR containing your application
$ mvn package
...
[INFO] Building jar: {..}/{..}/target/simple-project-1.0.jar# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \--class "SimpleApp" \--master local[4] \target/simple-project-1.0.jar
...
Lines with a: 72, lines with b: 39

我打包复制到YOUR_SPARK_HOME下了，执行的

bin/spark-submit.cmd --class "SimpleApp" --master local[4] simple-project-1.0.jar

从这里去哪里

恭喜您运行了您的第一个 Spark 应用程序！

有关 API 的深入概述，请从 RDD 编程指南和 SQL 编程指南开始，或者查看其他组件的“编程指南”菜单。
要在集群上运行应用程序，请前往部署概述。
最后，Spark 在目录中包含几个示例（Scala、Java、Python、R）。您可以按如下方式运行它们：examples

# For Scala and Java, use run-example:
./bin/run-example SparkPi# For Python examples, use spark-submit directly:
./bin/spark-submit examples/src/main/python/pi.py# For R examples, use spark-submit directly:
./bin/spark-submit examples/src/main/r/dataframe.R

配置参数 - Spark 3.5.5 Documentation

idea中使用

SparkWordCount

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;public class SparkWordCount {public static void main(String[] args) {System.setProperty("hadoop.home.dir", "D:\\hadoop-3.3.6\\hadoop-3.3.6");SparkConf sparkConf = new SparkConf().set("spark.testing.memory", "1024000000");String logFile = "D:\\work\\demo\\HELP.md"; // Should be some file on your systemSparkSession spark = SparkSession.builder().appName("Simple Application").master("local[4]").config(sparkConf).getOrCreate();Dataset<String> logData = spark.read().textFile(logFile).cache();long numAs = logData.filter((FilterFunction<String>) s -> s.contains("a")).count();long numBs = logData.filter((FilterFunction<String>) s -> s.contains("b")).count();System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);spark.stop();}
}
//输出：Lines with a: 15, lines with b: 5

pom

<!-- Spark Core 依赖（Java 必须要有） --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.5.5</version></dependency><!-- Spark SQL（如果你使用 Spark SQL，可以加上） --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>3.5.5</version></dependency><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.87.Final</version></dependency>

在这里插入图片描述

可能会遇到下边问题

问题1

Could not locate executable null\bin\winutils.exe in the Hadoop binaries.
下载hadoop https://www.apache.org/dyn/closer.cgi/hadoop/common/hadoop-3.3.6/hadoop-3.3.6.tar.gz 配置环境变量，下载https://github.com/cdarlint/winutils/archive/refs/heads/master.zip放入hadoop的bin目录
代码写上里边加上，
```
System.setProperty("hadoop.home.dir", "D:\\hadoop-3.3.6\\hadoop-3.3.6");
```
D:\hadoop-3.3.6\hadoop-3.3.6是我的hadoop目录你自己写上你的

问题2

cmd执行
```
hadoop version
```
如果报错
Error: JAVA_HOME is incorrectly set.
Please update D:\hadoop-3.3.6\hadoop-3.3.6\etc\hadoop\hadoop-env.cmd
这里提示的D:\hadoop-3.3.6\hadoop-3.3.6\etc\hadoop\hadoop-env.cmd是我的hadoop目录的这个文件，你的按你的提示的改对应地方。可能是jdk有空格

hadoop-env.cmd中修改

set JAVA_HOME=%JAVA_HOME%

改为

set JAVA_HOME=C:\PROGRA~2\Java\jdk1.8.0_102

Program Files和Program Files (x86)这两个目录分别表示为：PROGRA~1 和 PROGRA~2
C:\Program Files (x86)\Java\jdk1.8.0_102是我的jdk路径，你改成你的，记得最后别像网上说的乱改jdk的目录，别搞得你环境都不对了，像我这样改。

问题3

[INVALID_DRIVER_MEMORY] System memory 259522560 must be at least 471859200. Please increase heap size using the --driver-memory option or “spark.driver.memory” in Spark configuration.
在代码中加上系统属性

SparkConf sparkConf = new SparkConf().set("spark.testing.memory", "102400000");设置大于450M

问题4

如果你的netty跟这个冲突也可能报错：Exception in thread “main” java.lang.NoSuchMethodError: io.netty.buffer.PooledByteBufAllocator.(ZIIIIIIZ)V
pom中加上，这里主要是看你spark用的什么版本的，在这看，我的spark安装目录是D:\spark\spark-3.4.4-bin-hadoop3取下边的jars目录找netty看看它用的什么jar
```
D:\spark\spark-3.4.4-bin-hadoop3\jars
```
然后自己pom跟他用的不一样会出现这个问题，记得改成和他一样的。或者你找他版本和你netty一样的spark包部署，就是你兼容它或者让它然兼容你
```
<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.87.Final</version></dependency>
```

 #无shuftesc.makeRDD(List(1,2,3,4)).map(num=>num).collect#有Shufflesc.makeRDD(List(1,2,3,4)).groupBy(_%2).collect


Job Id ▾	Description	Submitted	Duration	Stages: Succeeded/Total	Tasks (for all stages): Succeeded/Total
1	collect at :24collect at :24	2025/03/26 10:30:34	1 s	2/2	16/16
0	collect at :24collect at :24	2025/03/26 10:22:17	0.4 s	1/1	8/8

需要文件中转所以变为16，乘2了，shuffle操作分区越多消耗越大

Stage Id	Description	Submitted	Duration	Tasks: Succeeded/Total	Input	Output	Shuffle Read ▾	Shuffle Write
2	collect at :24+details	2025/03/26 10:30:35	94 ms	8/8			208.0 B
1	groupBy at :24+details	2025/03/26 10:30:34	0.8 s	8/8				208.0 B

优化shufte，可以改分区，例如上边的例子数据就2个分区可以直接指定2个分区

sc.makeRDD(List(1,2,3,4)).groupBy(_%2,2).collect

Job Id ▾	Description	Submitted	Duration	Stages: Succeeded/Total	Tasks (for all stages): Succeeded/Total
2	collect at :24collect at :24	2025/03/26 10:46:19	44 ms	2/2	10/10

Stage Id ▾	Description	Submitted	Duration	Tasks: Succeeded/Total	Input	Output	Shuffle Read	Shuffle Write
4	collect at :24+details	2025/03/26 10:46:19	3.1 min	2/2			208.0 B
3	groupBy at :24+details	2025/03/26 10:46:19	34 ms	8/8				208.0 B

发现分区少了，消耗的性能降低了

血缘关系rdd.toDebugString

依赖关系rdd.dependencies

Apache Spark - 用于大规模数据分析的统一引擎

Apache Spark - 用于大规模数据分析的统一引擎

下载

运行示例和 Shell

使用 Spark Connect 在 Anywhere 上运行 Spark 客户端应用程序

在集群上启动

从这里去哪里

使用 Spark Shell 进行交互式分析

基本

有关数据集作的更多信息

缓存

自包含应用程序

从这里去哪里

idea中使用

可能会遇到下边问题

问题1

问题2

问题3

问题4

相关资讯

热文排行

最新新闻

推荐新闻

热搜词