[译]Spark快速开始

2023年3月22日 228次阅读 来源: steanxy

本文是一个如何使用Spark的简要教程。首先通过Spark的交互式Shell来介绍API（使用Python或Scala），然后展示如何用Java，Scala和Python来写Spark引用程序。更完整的内容请参考编程指南。

为了跟随这篇指南进行学习，请首先从Spark网站上下载一个Spark release包。因为我们这里不会使用HDFS，所以下载一个用于任意Hadoop版本的包即可。

使用Spark Shell进行交互式分析

基础

Spark的Shell提供了一种学习API的简单方法，同样也是交互式分析数据的强大工具。在Scala（运行在Java虚拟机上，可以使用现有的Java库）或Python中可用。在Spark目录下运行下面命令：

./bin/spark-shell

Spark的主要抽象是弹性分布式数据集（RDD）。RDD可以通过Hadoop输入格式（比如HDFS文件）或者其它RDD转换来创建。让我们使用Spark源路径的README文件的文本内容来创建一个RDD。

scala> val textFile = sc.textFile("README.md")
textFile: org.apache.spark.rdd.RDD[String] = README.md MapPartitionsRDD[1] at textFile at <console>:25

RDD有可以返回值的actions，有可以返回指向新RDD指针的transformations。让我们从几个actions开始：

scala> textFile.count() // Number of items in this RDD
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 RDD
res1: String = # Apache Spark

接下来使用transformation。我们使用filter transformation来返回一个包含文件条目子集的RDD。

scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:27

可以把transformations和actions链接起来使用。

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

更多RDD操作

RDD的actions和transformations可用于更复杂的计算。如我们想找到单词最多的行：

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

首先使用map函数将每一行转换成一个整数，创建了一个新RDD。然后在这个新RDD上调用reduce函数找出最大行的单词数。map和reduce的参数是Scala函数式（闭包），并且可以使用任何语言特性或者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(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[8] at reduceByKey at <console>:28

这里我们结合了flatmap，map和reduceByKey转换来计算文件中每个单词的数量，生成一个(String，Int)对的RDD。
为了在我们的Shell中统计单词数量，可以使用collect action：

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 = MapPartitionsRDD[2] at filter at <console>:27

scala> linesWithSpark.count()
res8: Long = 15

scala> linesWithSpark.count()
res9: Long = 15

用Spark来探索和缓存一个100行的文本文件看起来有点呆。有意思的是这些相同的函数可以用在非常庞大的数据集上，即使跨越数十或者数百个节点。你也可以像编程指南中描述的一样通过使用bin/spark-shell连接到集群的方式，来交互式地做这件事。

独立的应用程序

假设我们想用Spark API写一个独立的应用程序。可以使用Scala（with sbt），Java（with Maven）和Python实现一个简单的应用程序。

在Scala中创建一个简单的Spark应用程序，命名为SimpleApp.scala。

/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
    val conf = new SparkConf().setAppName("Simple Application")
    val sc = new SparkContext(conf)
    val logData = sc.textFile(logFile, 2).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")
    sc.stop()
  }
}

可以看到应用程序应该定义main()函数而不是扩展scala.App。使用scala.App的子类可能会运行不正常。这个程序就计算了Spark的README文件中包含a的行数和包含b的行数。注意把YOUR_SPARK_HOME替换成Spark的安装路径。不像之前使用Spark Shell的例子会初始化自己的SparkContext，我们需要初始化SparkContext作为程序的一部分。

我们要传一个包含了应用程序信息的SparkConf对象给SparkContext构造函数。

应用程序需要依赖Spark API，所以会包含一个sbt配置文件，simple.sbt，里面说明了Spark是一个依赖。这个文件也添加了一个Spark依赖的库：

name := "Simple Project"

version := "1.0"

scalaVersion := "2.11.7"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"

为了让sbt正确运行，我们需要根据经典的目录结构来组织 SimpleApp.scala和simple.sbt。完成之后，我们可以创建一个包含我们应用程序代码的JAR包，然后使用spark-submit脚本来运行我们的程序。

# Your directory layout should look like this
$ find .
.
./simple.sbt
./src
./src/main
./src/main/scala
./src/main/scala/SimpleApp.scala

# Package a jar containing your application
$ sbt package
...

 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

从这里开始

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

  • 查看API的深入概述，从Spark编程指南开始，或者看其它组件的编程指南菜单。
  • 要在集群上运行应用程序，请看部署概述。
  • 最后，Spark在examples目录中包含了很多示例（Scala，Java，Python，R），可以使用如下方式运行：
# 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
    原文作者：steanxy
    原文地址: https://www.jianshu.com/p/65e75de7f320
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞