版权声明：本文为博主皮皮http://blog.csdn.net/pipisorry原创文章，未经博主允许不得转载。 https://blog.csdn.net/pipisorry/article/details/52366356
http://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，或运行在本地模式时，使用专用字符串 “local”。在实践中，当应用程序运行在一个集群上时，你并不想要把 master 硬编码到你的程序中，你可以用 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'>)]
shell中初始化
$ ./bin/spark-shell --master local[4]
$ ./bin/spark-shell --master local[4] --jars code.jar
执行 spark-shell --help 获取完整的选项列表。在这之后，调用 spark-shell 会比 spark-submit 脚本更为普遍。
SQLContext
操作sql相关函数的类，需要这样才能使用：
sc = SparkContext(master='local[8]', appName='kmeans')
sql_ctx = SQLContext(sc)
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().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.
| Parameters: | blocking – Whether to block until unpersisting has completed |
|---|
- value
Return the broadcasted value
一个广播变量可以通过调用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功能]
累加器
顾名思义，累加器是一种只能通过关联操作进行“加”操作的变量，因此它能够高效的应用于并行操作中。它们能够用来实现counters和sums。Spark原生支持数值类型的累加器，开发者可以自己添加支持的类型。如果创建了一个具名的累加器，它可以在spark的UI中显示。这对于理解运行阶段(running stages)的过程有很重要的作用。（注意：这在python中还不被支持）
皮皮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机器学习示例代码]
皮皮blog
Spark编程实例
回归模型
[极简 Spark 入门笔记—第一个回归程序]
from:
ref: [Spark quick-start]
[针对Scala、 Java和Python的Spark programming-guide]
[王家林的spark视频公开课]
Machine Learning with Spark*