Apache Spark 2.2.0 中文文档 - 快速入门 | ApacheCN

./bin/spark-shell


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

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

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

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

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

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

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

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: Long = 15

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

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

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

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()
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:

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

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

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

注意, 这个应用程序我们应该定义一个 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:

name := "Simple Project"version := "1.0"scalaVersion := "2.11.8"libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.2.0"

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

# Your directory layout should look like this
$ find .
.
./build.sbt
./src
./src/main
./src/main/scala
./src/main/scala/SimpleApp.scala# Package a jar containing your application
$ sbt package
...
[info] Packaging {..}/{..}/target/scala-2.11/simple-project_2.11-1.0.jar# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit \
--class "SimpleApp" \
--master local[4] \
target/scala-2.11/simple-project_2.11-1.0.jar
...
Lines with a: 46, Lines with b: 23