本片简单的介绍如何使用Spark，我们将首先通过Spark的交互式shell（在Python或Scala中）介绍API。然后介绍如何使用Scale 和 python 编写应用程序。
首先从Spark网站下载Spark的打包版本
使用Spark shell 进行交互式分析
基本
Spark的shell提供了一种学习API的简单方法，以及一种以交互方式分析数据的强大工具。它可以在Scala（在Java VM上运行，因此是使用现有Java库的好方法）或Python中使用。通过在Spark目录中运行以下命令来启动它：
Scale
./bin/spark-shell
Spark的主要抽象是一个名为Dataset的分布式项目集合。可以从Hadoop InputFormats（例如HDFS文件）或通过转换其他数据集来创建数据集。让我们从Spark源目录中的README文件的文本中创建一个新的数据集：
scala> val textFile = spark.read.textFile("README.md")
textFile: org.apache.spark.sql.Dataset[String] = [value: string]
可以通过调用某些操作直接从Dataset获取值，或者转换数据集以获取新值。
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
现在让我们将这个数据集转换为新数据集。我们调用filter返回一个新的数据集，其中包含文件中的项目子集。
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: Long = 15
这首先将一行映射为整数值，从而创建一个新的数据集。reduce在该数据集上调用以查找最大字数。参数map和reduceScala函数文字（闭包），并可以使用任何语言功能或Scala / Java库。例如，我们可以轻松调用其他地方声明的函数。我们将使用Math.max()函数使这段代码更容易理解：
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
一种常见的数据流模式是MapReduce，由Hadoop推广。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将行数据集转换为单词数据集，然后组合groupByKey并count计算文件中的单词计数作为（字符串，长整数）对的数据集。要在我们的shell中收集单词count，我们可以调用collect：
scala> wordCounts.collect()
res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)
缓存
Spark还支持将数据集提取到群集范围的内存缓存中。这在重复访问数据时非常有用，例如查询小的“热”数据集或运行像PageRank这样的迭代算法时。举个简单的例子，让我们标记linesWithSpark要缓存的数据集：
scala> linesWithSpark.cache()
res7: linesWithSpark.type = [value: string]
scala> linesWithSpark.count()
res8: Long = 15
scala> linesWithSpark.count()
res9: Long = 15
用Spark来探索和缓存100行文本文件似乎很愚蠢。有趣的是，这些相同的功能可用于非常大的数据集，即使它们跨越数十个或数百个节点进行条带化。
自含的应用程序
假设我们希望使用Spark API编写一个自包含的应用程序。我们将在Scala
我们将在Scala中创建一个非常简单的Spark应用程序 - 事实上，它的名字很简单SimpleApp.scala：
/* SimpleApp.scala */
import org.apache.spark.sql.SparkSession
object 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可能无法正常工作。
