快速入门

本教程提供了如何使用 Spark 的快速入门介绍。首先通过运行 Spark 交互式的 shell（在 Python 或 Scala 中）来介绍 API，然后展示如何使用 Java，Scala 和 Python 来编写应用程序。

为了继续阅读本指南，首先从 Spark 官网 下载 Spark 的发行包。因为我们不使用 HDFS，所以你可以下载一个任何 Hadoop 版本的软件包。

请注意，在 Spark 2.0 之前，Spark 的主要编程接口是弹性分布式数据集（RDD）。 在 Spark 2.0 之后，RDD 被 Dataset 替换，它是像RDD 一样的 strongly-typed（强类型），但是在引擎盖下更加优化。 RDD 接口仍然受支持，您可以在 RDD 编程指南 中获得更完整的参考。 但是，我们强烈建议您切换到使用 Dataset（数据集），其性能要更优于 RDD。 请参阅 SQL 编程指南 获取更多有关 Dataset 的信息。

安全

默认情况下，Spark中的安全性处于关闭状态。这意味着您默认情况下容易受到攻击。在下载和运行Spark之前，请参阅Spark Security。

使用 Spark Shell 进行交互式分析

基础

Spark shell 提供了一种来学习该 API 比较简单的方式，以及一个强大的来分析数据交互的工具。在 Scala（运行于 Java 虚拟机之上，并能很好的调用已存在的 Java 类库）或者 Python 中它是可用的。通过在 Spark 目录中运行以下的命令来启动它:

  1. ./bin/spark-shell

Spark 的主要抽象是一个称为 Dataset 的分布式的 item 集合。Datasets 可以从 Hadoop 的 InputFormats（例如 HDFS文件）或者通过其它的 Datasets 转换来创建。让我们从 Spark 源目录中的 README 文件来创建一个新的 Dataset:

  1. scala> val textFile = spark.read.textFile("README.md")
  2. textFile: org.apache.spark.sql.Dataset[String] = [value: string]

您可以直接从 Dataset 中获取 values（值），通过调用一些 actions（动作），或者 transform（转换）Dataset 以获得一个新的。更多细节，请参阅 API doc。

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

现在让我们 transform 这个 Dataset 以获得一个新的 。我们调用 filter 以返回一个新的 Dataset，它是文件中的 items 的一个子集。

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

我们可以链式操作 transformation（转换）和 action（动作）:

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

Spark有一个主要的抽象概念叫做 Dataset 的分布式集合类。Dataset 可以从Hadoop InputFormats（例如HDFS文件）或通过 transforming 其他数据集来创建数据集。 由于Python的动态特性，我们不需要在Python中定义强类型的 Dataset。 因此，Python中的所有数据集都是 Dataset[Row]，我们称之为”DataFrame” 来与 Pandas 和R中的数据框概念一致。让我们从Spark源文件中的README文件中创建一个新的 DataFrame 目录：

  1. >>> textFile = spark.read.text("README.md")

您可以通过调用某些action直接从DataFrame获取值，也可以transform DataFrame以获取新的DataFrame。 有关详细信息，请阅读 API doc.

  1. >>> textFile.count()  # 这个DataFrame 有多少行
  2. 126
  4. >>> textFile.first()  # DataFrame的第一行
  5. Row(value=u'# Apache Spark')

现在让我们 transform（转换） 这个DataFrame来获得一个新的DataFrame. 我们调用 filter 方法来返回文件中的一个子集.

  1. >>> linesWithSpark = textFile.filter(textFile.value.contains("Spark"))

我们可以把 transform 和 acition 连在一起用:

  1. >>> textFile.filter(textFile.value.contains("Spark")).count()  # 统计文件中 "Spark" 字符串有多少个
  2. 15

Dataset 上的更多操作

Dataset actions（操作）和 transformations（转换）可以用于更复杂的计算。例如，统计出现次数最多的行 :

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

第一个 map 操作创建一个新的 Dataset，将一行数据 map 为一个整型值。在 Dataset 上调用 reduce 来找到最大的行计数。参数 map 与 reduce 是 Scala 函数（closures），并且可以使用 Scala/Java 库的任何语言特性。例如，我们可以很容易地调用函数声明，我们将定义一个 max 函数来使代码更易于理解 :

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

一种常见的数据流模式是被 Hadoop 所推广的 MapReduce。Spark 可以很容易实现 MapReduce:

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

在这里，我们调用了 flatMap 以 transform 一个 lines 的 Dataset 为一个 words 的 Dataset，然后结合 groupByKey 和 count 来计算文件中每个单词的 counts 作为一个 (String, Long) 的 Dataset pairs。要在 shell 中收集 word counts，我们可以调用 collect:

  1. scala> wordCounts.collect()
  2. res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)
  1. >>> from pyspark.sql.functions import *
  2. >>> textFile.select(size(split(textFile.value, "\s+")).name("numWords")).agg(max(col("numWords"))).collect()
  3. [Row(max(numWords)=15)]

这首先将一行映射为一个整数值并且别名为 “numWords” ，从中创建一个新的DataFrame. 在该 DataFrame 上调用 agg 函数是为了找到最大词数（word count） . select 和 agg 的参数都是 Column 里的,我们也可以通过 df.colName 来获得该 DataFrame 的一列. 我们也可以导入 pyspark.sql.functions 来提供了许多方便的功能来从旧的列构建一个新的列,

一个常见的数据流模式是MapReduce，由Hadoop推广。 Spark可以轻松实现MapReduce流程：

  1. >>> wordCounts = textFile.select(explode(split(textFile.value, "\s+")).as("word")).groupBy("word").count()

在这里我们在 select 中使用 explode 函数来将一个 Dataset 的所有行转换成一个词的数据集，然后组合使用 groupBy 和 count 来计算文件中各个单词的计数作为 DataFrame 的两列: “word” 和 “count”.要在我们的shell中统计单词的词频，我们可以调用 collect：

  1. >>> wordCounts.collect()
  2. [Row(word=u'online', count=1), Row(word=u'graphs', count=1), ...]

缓存

Spark 还支持 Pulling（拉取）数据集到一个群集范围的内存缓存中。例如当查询一个小的 “hot” 数据集或运行一个像 PageRANK 这样的迭代算法时，在数据被重复访问时是非常高效的。举一个简单的例子，让我们标记我们的 linesWithSpark 数据集到缓存中:

  1. scala> linesWithSpark.cache()
  2. res7: linesWithSpark.type = [value: string]
  4. scala> linesWithSpark.count()
  5. res8: Long = 15
  7. scala> linesWithSpark.count()
  8. res9: Long = 15

使用 Spark 来探索和缓存一个 100 行的文本文件看起来比较愚蠢。有趣的是，即使在他们跨越几十或者几百个节点时，这些相同的函数也可以用于非常大的数据集。您也可以像 编程指南. 中描述的一样通过连接 bin/spark-shell 到集群中，使用交互式的方式来做这件事情。

  1. >>> linesWithSpark.cache()
  3. >>> linesWithSpark.count()
  4. 15
  6. >>> linesWithSpark.count()
  7. 15

使用Spark探索和缓存100行文本文件似乎很愚蠢. T有趣的是，这些相同的功能可用于非常大的数据集，即使它们跨越数十个或数百个节点交错着, 你也可以通过bin/pyspark 连接到集群来进行交互，详细描述在 RDD programming guide.

独立的应用

假设我们希望使用 Spark API 来创建一个独立的应用程序。我们在 Scala（SBT），Java（Maven）和 Python 中练习一个简单应用程序。

我们将在 Scala 中创建一个非常简单的 Spark 应用程序 - 很简单的，事实上，它名为 SimpleApp.scala:

  1. /* SimpleApp.scala */
  2. import org.apache.spark.sql.SparkSession
  4. object SimpleApp {
  5.   def main(args: Array[String]) {
  6.     val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
  7.     val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
  8.     val logData = spark.read.textFile(logFile).cache()
  9.     val numAs = logData.filter(line => line.contains("a")).count()
  10.     val numBs = logData.filter(line => line.contains("b")).count()
  11.     println(s"Lines with a: $numAs, Lines with b: $numBs")
  12.     spark.stop()
  13.   }
  14. }

注意，这个应用程序我们应该定义一个 main() 方法而不是去扩展 scala.App。使用 scala.App 的子类可能不会正常运行。

该程序仅仅统计了 Spark README 文件中每一行包含 ‘a’ 的数量和包含 ‘b’ 的数量。注意，您需要将 YOUR_SPARK_HOME 替换为您 Spark 安装的位置。不像先前使用 spark shell 操作的示例，它们初始化了它们自己的 SparkContext，我们初始化了一个 SparkContext 作为应用程序的一部分。

我们调用 SparkSession.builder 以构造一个 [[SparkSession]]，然后设置 application name（应用名称），最终调用 getOrCreate 以获得 [[SparkSession]] 实例。

我们的应用依赖了 Spark API，所以我们将包含一个名为 build.sbt 的 sbt 配置文件，它描述了 Spark 的依赖。该文件也会添加一个 Spark 依赖的 repository:

  1. name := "Simple Project"
  3. version := "1.0"
  5. scalaVersion := "2.12.8"
  7. libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.4.4"

为了让 sbt 正常的运行，我们需要根据经典的目录结构来布局 SimpleApp.scala 和 build.sbt 文件。在成功后，我们可以创建一个包含应用程序代码的 JAR 包，然后使用 spark-submit 脚本来运行我们的程序。

  1. # Your directory layout should look like this
  2. $ find .
  3. .
  4. ./build.sbt
  5. ./src
  6. ./src/main
  7. ./src/main/scala
  8. ./src/main/scala/SimpleApp.scala
  10. # Package a jar containing your application
  11. $ sbt package
  12. ...
  13. [info] Packaging {..}/{..}/target/scala-2.12/simple-project_2.12-1.0.jar
  15. # Use spark-submit to run your application
  16. $ YOUR_SPARK_HOME/bin/spark-submit \
  17.   --class "SimpleApp" \
  18.   --master local[4] \
  19.   target/scala-2.12/simple-project_2.12-1.0.jar
  20. ...
  21. Lines with a: 46, Lines with b: 23

这个例子使用Maven来编译成一个jar应用程序，其他的构建系统（如Ant、Gradle，译者注）也可以。

我们会创建一个非常简单的Spark应用，SimpleApp.java:

  1. /* SimpleApp.java */
  2. import org.apache.spark.sql.SparkSession;
  4. public class SimpleApp {
  5.   public static void main(String[] args) {
  6.     String logFile = "YOUR_SPARK_HOME/README.md"; // Should be some file on your system
  7.     SparkSession spark = SparkSession.builder().appName("Simple Application").getOrCreate();
  8.     Dataset<String> logData = spark.read.textFile(logFile).cache();
  10.     long numAs = logData.filter(s -> s.contains("a")).count();
  11.     long numBs = logData.filter(s -> s.contains("b")).count();
  13.     System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);
  15.     spark.stop();
  16.   }
  17. }

这个程序计算Spark README文档中包含字母’a’和字母’b’的行数。注意把YOUR_SPARK_HOME修改成你的Spark的安装目录。 跟之前的Spark shell不同，我们需要初始化SparkSession。

把Spark依赖添加到Maven的pom.xml文件里。 注意Spark的artifacts使用Scala版本进行标记。

  1. <project>
  2.   <groupId>edu.berkeley</groupId>
  3.   <artifactId>simple-project</artifactId>
  4.   <modelVersion>4.0.0</modelVersion>
  5.   <name>Simple Project</name>
  6.   <packaging>jar</packaging>
  7.   <version>1.0</version>
  8.   <dependencies>
  9.     <dependency> <!-- Spark dependency -->
  10.       <groupId>org.apache.spark</groupId>
  11.       <artifactId>spark-sql_2.12</artifactId>
  12.       <version>2.4.4</version>
  13.       <scope>provided</scope>
  14.     </dependency>
  15.   </dependencies>
  16. </project>

我们按照Maven经典的目录结构组织这些文件：

  1. $ find .
  2. ./pom.xml
  3. ./src
  4. ./src/main
  5. ./src/main/java
  6. ./src/main/java/SimpleApp.java

现在我们用Maven打包这个应用，然后用./bin/spark-submit执行它。

  1. # 打包包含应用程序的JAR
  2. $ mvn package
  3. ...
  4. [INFO] Building jar: {..}/{..}/target/simple-project-1.0.jar
  6. # 用spark-submit来运行程序
  7. $ YOUR_SPARK_HOME/bin/spark-submit \
  8.   --class "SimpleApp" \
  9.   --master local[4] \
  10.   target/simple-project-1.0.jar
  11. ...
  12. Lines with a: 46, Lines with b: 23

现在我们来展示如何用python API 来写一个应用 (pyspark).

如果要构建打包的PySpark应用程序或库，则可以添加以下内容到setup.py文件中：

  1.  install_requires=[
  2.         'pyspark=={site.SPARK_VERSION}'
  3.     ]

我们以一个简单的例子为例，创建一个简单的pyspark 应用 SimpleApp.py:

  1. """SimpleApp.py"""
  2. from pyspark.sql import SparkSession
  4. logFile = "YOUR_SPARK_HOME/README.md"  # Should be some file on your system
  5. spark = SparkSession.builder().appName(appName).master(master).getOrCreate()
  6. logData = spark.read.text(logFile).cache()
  8. numAs = logData.filter(logData.value.contains('a')).count()
  9. numBs = logData.filter(logData.value.contains('b')).count()
  11. print("Lines with a: %i, lines with b: %i" % (numAs, numBs))
  13. spark.stop()

该程序只是统计计算在该文本中包含a字母和包含b字母的行数. 请注意你需要将 YOUR_SPARK_HOME 替换成你的spark路径.就像scala 示例和java示例一样，我们使用 SparkSession 来创建数据集， 对于使用自定义类护着第三方库的应用程序，我们还可以通过 spark-submit 带着 --py-files 来添加代码依赖 , 我们也可以通过把代码打成zip包来进行依赖添加 (详细请看 spark-submit --help ). SimpleApp 是个简单的例子我们不需要添加特别的代码或自定义类.

我们可以通过 bin/spark-submit 脚本来运行应用:

  1. # Use spark-submit to run your application
  2. $ YOUR_SPARK_HOME/bin/spark-submit \
  3.   --master local[4] \
  4.   SimpleApp.py
  5. ...
  6. Lines with a: 46, Lines with b: 23

如果您的环境中已安装PySpark pip（例如pip install pyspark），则可以使用常规Python解释器运行应用程序，也可以根据需要使用前面的“ spark-submit”。

  1. # Use the Python interpreter to run your application
  2. $ python SimpleApp.py
  3. ...
  4. Lines with a: 46, Lines with b: 23

快速跳转

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

  1. # 针对 Scala 和 Java，使用 run-example:
  2. ./bin/run-example SparkPi
  4. # 针对 Python 示例，直接使用 spark-submit:
  5. ./bin/spark-submit examples/src/main/python/pi.py
  7. # 针对 R 示例，直接使用 spark-submit:
  9. ./bin/spark-submit examples/src/main/r/dataframe.R