当前位置：首页 > news >正文

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

news 来源：原创 2025/5/1 16:36:28

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

下载
运行示例和 Shell
- 使用 Spark Connect 在 Anywhere 上运行 Spark 客户端应用程序
在集群上启动
从这里去哪里
使用 Spark Shell 进行交互式分析
- 基本
- 有关数据集作的更多信息
- 缓存
自包含应用程序
从这里去哪里

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 outputs

scala> 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.Math

scala> 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 = 15

scala> 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 system
    SparkSession 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