Spark：Spark 编程模型及快速入门

た入场券 2022-07-17 03:39 136阅读 0赞

http://[blog.csdn.net/pipisorry/article/details/52366356  
][blog.csdn.net_pipisorry_article_details_52366356]

# Spark编程模型 #

## SparkContext类和SparkConf类 ##

### 代码中初始化    ###

我们可通过如下方式调用 SparkContext 的简单构造函数,以默认的参数值来创建相应的对象。  
val sc = new SparkContext("local\[4\]", "Test Spark App")

这段代码会创建一个4线程的 SparkContext 对象 。

appName 参数是你程序的名字，它会显示在 cluster UI 上，上面代码相应的任务命名为 Test Spark APP。

master 是 [Spark, Mesos 或 YARN 集群的 URL][Spark_ Mesos _ YARN _ URL]，或运行在本地模式时，使用专用字符串 “local”。在实践中，当应用程序运行在一个集群上时，你并不想要把 master 硬编码到你的程序中，你可以[用 spark-submit 启动你的应用程序][spark-submit]的时候传递它。然而，你可以在本地测试和单元测试中使用 “local” 运行 Spark 进程。

\[[SparkContext][]\]

\[pyspark.SparkContext(*master=None*, *appName=None*, *sparkHome=None*, *pyFiles=None*, *environment=None*, *batchSize=0*, *serializer=PickleSerializer()*, *conf=None*, *gateway=None*, *jsc=None*, *profiler\_cls=<class 'pyspark.profiler.BasicProfiler'>*)[¶][Link 1]\]

### shell中初始化 ###

在 Spark shell 中，有一个专有的 SparkContext 已经为你创建好。在变量中叫做 sc。你自己创建的 SparkContext 将无法工作。可以用 --master 参数来设置 SparkContext 要连接的集群，用 --jars 来设置需要添加到 classpath 中的 JAR 包，如果有多个 JAR 包使用逗号分割符连接它们。例如：在一个拥有 4 核的环境上运行 bin/spark-shell，使用：

$ ./bin/spark-shell --master local[4]

或在 classpath 中添加 code.jar，使用：

$ ./bin/spark-shell --master local[4] --jars code.jar

执行 spark-shell --help 获取完整的选项列表。在这之后，调用 spark-shell 会比 [spark-submit 脚本][spark-submit]更为普遍。

### SQLContext ###

操作sql相关函数的类，需要这样才能使用：

sc = SparkContext(master='local[8]', appName='kmeans')
    sql_ctx = SQLContext(sc)

## Spark：弹性分布式数据集RDD及spark转换和操作  ##

\[[Spark：弹性分布式数据集RDD及spark转换和操作 ][Spark_RDD_spark_]\]

# 共享变量 #

Spark 的共享变量能被运行在并行计算中。默认情况下，当 Spark 运行一个并行函数时，这个并行函数会作为一个任务集在不同的节点上运行，它会把函数里使用的每个变量都复制搬运到每个任务中。有时，一个变量需要被共享到交叉任务中或驱动程序和任务之间。

Spark 支持 2 种类型的共享变量：广播变量(broadcast variables)，用来在所有节点的内存中缓存一个值；累加器(accumulators)，仅仅只能执行“添加(added)”操作，例如：记数器(counters)和求和(sums)。

### 广播变量broadcast variable ###

class pyspark.Broadcast(sc=None, value=None, pickle\_registry=None, path=None)

A broadcast variable created with [SparkContext.broadcast()][SparkContext.broadcast].Access its value through value.

Examples:

>>> from pyspark.context import SparkContext
    >>> sc = SparkContext('local', 'test')
    >>> b = sc.broadcast([1, 2, 3, 4, 5])
    >>> b.value
    [1, 2, 3, 4, 5]
    >>> sc.parallelize([0, 0]).flatMap(lambda x: b.value).collect()
    [1, 2, 3, 4, 5, 1, 2, 3, 4, 5]
    >>> b.unpersist()

>>> large_broadcast = sc.broadcast(range(10000))

destroy()

Destroy all data and metadata related to this broadcast variable.Use this with caution; once a broadcast variable has been destroyed,it cannot be used again. This method blocks until destroy hascompleted.

dump(value, f)

load(path)

unpersist(blocking=False)

Delete cached copies of this broadcast on the executors. If thebroadcast is used after this is called, it will need to bere-sent to each executor.

<table>     
 <tbody> 
  <tr> 
   <th>Parameters:</th> 
   <td><strong>blocking</strong> – Whether to block until unpersisting has completed</td> 
  </tr> 
 </tbody> 
</table>

value

Return the broadcasted value

广播变量允许程序员缓存一个只读的变量在每台机器上面，而不是每个任务保存一份拷贝，它由运行 SparkContext 的驱动程序创建后发送给会参与计算的节点。例如，利用广播变量，我们能够以一种更有效率的方式将一个大数据量输入集合的副本分配给每个节点。对那些需要让各工作节点高效地访问相同数据的应用场景,比如机器学习,这非常有用。（Broadcast variables allow theprogrammer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks.They can be used, for example,to give every node a copy of a large input dataset in an efficient manner.）Spark也尝试着利用有效的广播算法去分配广播变量，以减少通信的成本。

一个广播变量可以通过调用SparkContext.broadcast(v)方法从一个初始变量v中创建。广播变量是v的一个包装变量，它的值可以通过value方法访问。

val broadcastAList = sc.broadcast(List("a", "b", "c", "d", "e"))  
终端的输出表明,广播变量存储在内存中。

广播变量也可以被非驱动程序所在的节点(即工作节点)访问,访问的方法是调用该变量的value 方法:  
sc.parallelize(List("1", "2", "3")).map(x => broadcastAList.value ++ x).collect

RDD分布                                  各节点map操作                                       合并到主节点

广播变量创建以后，我们就能够在集群的任何函数中使用它来代替变量v，这样我们就不需要再次传递变量v到每个节点上。另外，为了保证所有的节点得到广播变量具有相同的值，对象v不能在广播之后被修改。

\[[Broadcast功能][Broadcast]\]

### 累加器 ###

顾名思义，累加器是一种只能通过关联操作进行“加”操作的变量，因此它能够高效的应用于并行操作中。它们能够用来实现counters和sums。Spark原生支持数值类型的累加器，开发者可以自己添加支持的类型。如果创建了一个具名的累加器，它可以在spark的UI中显示。这对于理解运行阶段(running stages)的过程有很重要的作用。（注意：这在python中还不被支持）

[皮皮blog][blog]

# Spark编程入门 #

## Spark Scala 编程入门    ##

import org.apache.spark.SparkContext  
import org.apache.spark.SparkContext.\_  
/\*\*  
\* 用Scala编写的一个简单的Spark应用  
\*/  
object ScalaApp \{  
在主函数里,我们要初始化所需的 SparkContext 对象,并且用它通过 textFile 函数来访问CSV数据文件。之后对每一行原始字符串以逗号为分隔符进行分割,提取出相应的用户名、产品和价格信息,从而完成对原始文本的映射:  
def main(args: Array\[String\]) \{  
val sc = new SparkContext("local\[2\]", "First Spark App")  
// 将CSV格式的原始数据转化为(user,product,price)格式的记录集  
val data = sc.textFile("data/UserPurchaseHistory.csv")  
.map(line => line.split(","))  
.map(purchaseRecord => (purchaseRecord(0), purchaseRecord(1),  
purchaseRecord(2)))  
现在,我们有了一个RDD,其每条记录都由 (user, product, price) 三个字段构成。我们可以对商店计算如下指标:购买总次数 客户总个数 总收入1.4 最畅销的产品  
// 求购买次数  
val numPurchases = data.count()  
// 求有多少个不同客户购买过商品  
val uniqueUsers = data.map\{ case (user, product, price) => user \}.distinct().count()  
// 求和得出总收入  
val totalRevenue = data.map\{ case (user, product, price) => price.toDouble \}.sum()  
// 求最畅销的产品是什么  
val productsByPopularity = data  
.map\{ case (user, product, price) => (product, 1) \}  
.reduceByKey(\_ + \_)  
.collect()  
.sortBy(-\_.\_2)  
val mostPopular = productsByPopularity(0)  
最后那段计算最畅销产品的代码演示了如何进行Map/Reduce模式的计算,该模式随Hadoop而流行。第一步,我们将 (user, product, price) 格式的记录映射为 (product, 1) 格式。然后,我们执行一个 reduceByKey 操作,它会对各个产品的1值进行求和。转换后的RDD包含各个商品的购买次数。有了这个RDD后,我们可以调用 collect 函数,这会将其计算结果以Scala集合的形式返回驱动程序。之后在驱动程序的本地对这些记录按照购买次数进行排序。(注意,在实际处理大量数据时,我们通常通过 sortByKey 这类操作来对其进行并行排序。) 最后,可在终端上打印出计算结果:  
println("Total purchases: " + numPurchases)  
println("Unique users: " + uniqueUsers)  
println("Total revenue: " + totalRevenue)  
println("Most popular product: %s with %d purchases".  
format(mostPopular.\_1, mostPopular.\_2))  
\}  
\}  
可以在项目的主目录下执行 sbt run 命令来运行这个程序。如果你使用了IDE的话,也可以从Scala IDE直接运行。最终的输出应该与下面的内容相似:  
...  
\[info\] Compiling 1 Scala source to ...  
\[info\] Running ScalaApp  
...  
14/01/30 10:54:40 INFO spark.SparkContext: Job finished: collect at  
ScalaApp.scala:25, took 0.045181 s  
Total purchases: 5  
Unique users: 4  
Total revenue: 39.91  
Most popular product: iPhone Cover with 2 purchases

## Spark Python 编程入门    ##

Spark的Python API几乎覆盖了所有Scala API所能提供的功能,但的确有些特性,比如SparkStreaming和个别的API方法 , 暂不支持 。具体可参见《 Spark 编 程 指 南 》的 Python 部分 :http://spark.apache.org/docs/latest/programming-guide.html。

"""用Python编写的一个简单Spark应用"""  
from pyspark import SparkContext  
sc = SparkContext("local\[2\]", "First Spark App")  
\# 将CSV格式的原始数据转化为(user,product,price)格式的记录集  
data = sc.textFile("data/UserPurchaseHistory.csv").map(lambda line:  
line.split(",")).map(lambda record: (record\[0\], record\[1\], record\[2\]))  
\# 求总购买次数  
numPurchases = data.count()  
\# 求有多少不同客户购买过商品  
uniqueUsers = data.map(lambda record: record\[0\]).distinct().count()  
\# 求和得出总收入  
totalRevenue = data.map(lambda record: float(record\[2\])).sum()  
\# 求最畅销的产品是什么  
products = data.map(lambda record: (record\[1\], 1.0)).  
reduceByKey(lambda a, b: a + b).collect()  
mostPopular = sorted(products, key=lambda x: x\[1\], reverse=True)\[0\]  
print("Total purchases: %d" % numPurchases)  
print("Unique users: %d" % uniqueUsers)  
print("Total revenue: %2.2f" % totalRevenue)  
print("Most popular product: %s with %d purchases" % (mostPopular\[0\], mostPopular\[1\]))

### 运行该脚本的最好方法是在脚本目录下运行如下命令 ###

>$SPARK\_HOME/bin/spark-submit pythonapp.py

添加到path后直接$ spark-submit SparkLDA.py

代码及数据下载\[[spark机器学习示例代码][spark]\]

[皮皮blog][blog]

## Spark编程实例 ##

回归模型

\[[极简 Spark 入门笔记—第一个回归程序][Spark]\]

from:

ref: \[[Spark quick-start][]\]  
\[针对Scala、 Java和Python的[Spark programming-guide][]\]

\[王家林的[spark视频公开课][spark 1]\]

Machine Learning with Spark\*

[blog.csdn.net_pipisorry_article_details_52366356]: http://blog.csdn.net/pipisorry/article/details/52366356
[Spark_ Mesos _ YARN _ URL]: https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
[spark-submit]: https://spark.apache.org/docs/latest/submitting-applications.html
[SparkContext]: http://spark.apache.org/docs/latest/api/scala/org/apache/spark/SparkContext.html
[Link 1]: http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.SparkContext
[Spark_RDD_spark_]: http://blog.csdn.net/pipisorry/article/details/53257188
[SparkContext.broadcast]: http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.SparkContext.broadcast
[Broadcast]: https://www.kancloud.cn/kancloud/spark-internals/45238
[blog]: http://blog.csdn.net/pipisorry
[spark]: http://download.csdn.net/detail/pipisorry/9454130
[Spark]: http://mp.weixin.qq.com/s?__biz=MjM5NDQ3NTkwMA==&mid=206344219&idx=1&sn=c875b6b1e72b87cde396316d39c574e4&scene=2&from=timeline&isappinstalled=0#rd
[Spark quick-start]: http://spark.apache.org/docs/latest/quick-start.html
[Spark programming-guide]: http://spark.apace.org/docs/latest/programming-guide.html
[spark 1]: http://edu.51cto.com/lesson/id-31786.html