Spark系列：Python版Spark编程指南
scala不是内部或外部命令
⽬录
⼀、介绍
⼆、连接Spark
三、创建RDD
四、RDD常⽤的转换 Transformation
五、RDD 常⽤的执⾏动作 Action
⼆、连接Spark
Spark1.3.0只⽀持Python2.6或更⾼的版本（但不⽀持Python3）。它使⽤了标准的CPython解释器，所以诸如NumPy⼀类的C库也是可以使⽤的。
通过Spark⽬录下的bin/spark-submit脚本你可以在Python中运⾏Spark应⽤。这个脚本会载⼊Spark的Java/Scala库然后让你将应⽤提交到集中。你可以执⾏bin/pyspark来打开Python的交互命令⾏。
如果你希望访问HDFS上的数据，你需要为你使⽤的HDFS版本建⽴⼀个PySpark连接。常见的HDFS版本标签都已经列在了这个第三⽅发⾏版页⾯。
最后，你需要将⼀些Spark的类import到你的程序中。加⼊如下这⾏：
from pyspark import SparkContext, SparkConf
在⼀个Spark程序中要做的第⼀件事就是创建⼀个SparkContext对象来告诉Spark如何连接⼀个集。为了创建SparkContext，你⾸先需要创建⼀个SparkConf对象，这个对象会包含你的应⽤的⼀些相关信息。
conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)
appName参数是在集UI上显⽰的你的应⽤的名称。master是⼀个Spark、Mesos或YARN集的URL，如果你在本地运⾏那么这个参数应该是特殊的”local”字符串。在实际使⽤中，当你在集中运⾏你的程序，你⼀般不会把master参数写死在代码中，⽽是通过⽤spark-submit运⾏程序来获得这个参数。但是，在本地测试以及单元测试时，你仍需要⾃⾏传⼊”local”来运⾏Spark程序。
三、创建RDD
Spark是以RDD概念为中⼼运⾏的。RDD是⼀个容错的、可以被并⾏操作的元素集合。创建⼀个RDD有两个⽅法：在你的驱动程序中并⾏化⼀个已经存在的集合；从外部存储系统中引⽤⼀个数据集，这个存储系统可以是⼀个共享⽂件系统，⽐如HDFS、HBase或任意提供了Hadoop输⼊格式的数据来源。
并⾏化集合
并⾏化集合是通过在驱动程序中⼀个现有的迭代器或集合上调⽤SparkContext的parallelize⽅法建⽴的。为了创建⼀个能够并⾏操作的分布数据集，集合中的元素都会被拷贝。⽐如，以下语句创建了⼀个包含1到5的并⾏化集合：
data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)
分布数据集（distData）被建⽴起来之后，就可以进⾏并⾏操作了。⽐如，我们可以调⽤duce(lambda a, b: a+b)来对元素进⾏叠加。在后⽂中我们会描述分布数据集上⽀持的操作。
并⾏集合的⼀个重要参数是将数据集划分成分⽚的数量。对每⼀个分⽚，Spark会在集中运⾏⼀个对
应的任务。典型情况下，集中的每⼀个CPU将对应运⾏2-4个分⽚。⼀般情况下，Spark会根据当前集的情况⾃⾏设定分⽚数量。但是，你也可以通过将第⼆个参数传递给parallelize⽅法(⽐如sc.parallelize(data, 10))来⼿动确定分⽚数量。注意：有些代码中会使⽤切⽚（slice，分⽚的同义词）这个术语来保持向下兼容性。
⼀个简单的⽰例：
import findspark
findspark.init()
from pyspark import SparkContext
from pyspark import SparkConf
conf = SparkConf().setAppName("miniProject").setMaster("local[*]")
OrCreate(conf)
rdd=sc.parallelize([1,2,3,4,5])
rdd1=rdd.map(lambda r:r+10) #map对每⼀个元素操作
llect())
外部数据集
PySpark可以通过Hadoop⽀持的外部数据源（包括本地⽂件系统、HDFS、 Cassandra、HBase、亚马逊S3等等）建⽴分布数据集。Spark⽀持⽂本⽂件、序列⽂件以及其他任何 Hadoop输⼊格式⽂件。
通过⽂本⽂件创建RDD要使⽤SparkContext的textFile⽅法。这个⽅法会使⽤⼀个⽂件的URI（或本地⽂件路径，hdfs://、s3n://这样的URI等等）然后读⼊这个⽂件建⽴⼀个⽂本⾏的集合。以下是⼀个例⼦：
>>> distFile = sc.textFile("")
建⽴完成后distFile上就可以调⽤数据集操作了。⽐如，我们可以调⽤map和reduce操作来叠加所有⽂本⾏的长度，代码如下：
distFile.map(lambda s: len(s)).reduce(lambda a, b: a + b)
在Spark中读⼊⽂件时有⼏点要注意：
如果使⽤了本地⽂件路径时，要保证在worker节点上这个⽂件也能够通过这个路径访问。这点可以通过将这个⽂件拷贝到所有worker上或者使⽤⽹络挂载的共享⽂件系统来解决。
包括textFile在内的所有基于⽂件的Spark读⼊⽅法，都⽀持将⽂件夹、压缩⽂件、包含通配符的路径作为参数。⽐如，以下代码都是合法的：
textFile("/my/directory")
textFile("/my/directory/*.txt")
textFile("/my/directory/*.gz")
textFile⽅法也可以传⼊第⼆个可选参数来控制⽂件的分⽚数量。默认情况下，Spark会为⽂件的每⼀个块（在HDFS中块的⼤⼩默认是64MB）创建⼀个分⽚。但是你也可以通过传⼊⼀个更⼤的值来要求Spark建⽴更多的分⽚。注意，分⽚的数量绝不能⼩于⽂件块的数量。
除了⽂本⽂件之外，Spark的Python API还⽀持多种其他数据格式：
SparkContext.wholeTextFiles能够读⼊包含多个⼩⽂本⽂件的⽬录，然后为每⼀个⽂件返回⼀个（⽂件名，内容)对。这是与textFile⽅法为每⼀个⽂本⾏返回⼀条记录相对应的。
RDD.saveAsPickleFile和SparkContext.pickleFile⽀持将RDD以串⾏化的Python对象格式存储起来。串⾏化的过程中会以默认10个⼀批的数量批量处理。
序列⽂件和其他Hadoop输⼊输出格式。
注意
这个特性⽬前仍处于试验阶段，被标记为Experimental，⽬前只适⽤于⾼级⽤户。这个特性在未来可能会被基于Spark SQL的读写⽀持所取代，因为Spark SQL是更好的⽅式。
可写类型⽀持
PySpark序列⽂件⽀持利⽤Java作为中介载⼊⼀个键值对RDD，将可写类型转化成Java的基本类型，然后使⽤ Pyrolite将java结果对象串⾏化。当将⼀个键值对RDD储存到⼀个序列⽂件中时PySpark将会运⾏上述过程的相反过程。⾸先将Python对象反串⾏化成Java对象，然后转化成可写类型。以下可写类型会⾃动转换：
| 可写类型 | Python类型 |
| Text | unicode str|
| IntWritable | int |
| FloatWritable | float |
| DoubleWritable | float |
| BooleanWritable | bool |
| BytesWritable | bytearray |
| NullWritable | None |
| MapWritable | dict |
数组是不能⾃动转换的。⽤户需要在读写时指定ArrayWritable的⼦类型.在读⼊的时候，默认的转换器会把⾃定义的ArrayWritable⼦类型转化成Java的Object[]，之后串⾏化成Python的元组。为了获得Python的array.array类型来使⽤主要类型的数组，⽤户需要⾃⾏指定转换器。
保存和读取序列⽂件
和⽂本⽂件类似，序列⽂件可以通过指定路径来保存与读取。键值类型都可以⾃⾏指定，但是对于标
准可写类型可以不指定。
>>> rdd = sc.parallelize(range(1, 4)).map(lambda x: (x, "a" * x ))
>>> rdd.saveAsSequenceFile("path/to/file")
>>> sorted(sc.sequenceFile("path/to/file").collect())
[(1, u'a'), (2, u'aa'), (3, u'aaa')]
保存和读取其他Hadoop输⼊输出格式
PySpark同样⽀持写⼊和读出其他Hadoop输⼊输出格式，包括’新’和’旧’两种Hadoop MapReduce API。如果有必要，⼀个Hadoop配置可以以Python字典的形式传⼊。以下是⼀个例⼦，使⽤了Elasticsearch ESInputFormat：
$ SPARK_CLASSPATH=/path/to/elasticsearch-hadoop.jar ./bin/pyspark
>>> conf = {"es.resource" : "index/type"}  # assume Elasticsearch is running on localhost defaults
>>> rdd = sc.newAPIHadoopRDD("org.EsInputFormat",\
"org.apache.hadoop.io.NullWritable", "org.LinkedMapWritable", conf=conf)
>>> rdd.first()        # the result is a MapWritable that is converted to a Python dict
(u'Elasticsearch ID',
{u'field1': True,
u'field2': u'Some Text',
u'field3': 12345})
四、RDD常⽤的转换 Transformation
RDD⽀持两类操作：转化操作，⽤于从已有的数据集转化产⽣新的数据集；启动操作，⽤于在计算结束后向驱动程序返回结果。举个例⼦，map是⼀个转化操作，可以将数据集中每⼀个元素传给⼀个函数，同时将计算结果作为⼀个新的RDD返回。另⼀⽅⾯，reduce操作是⼀个启动操作，能够使⽤某些函数来聚集计算RDD中所有的元素，并且向驱动程序返回最终结果（同时还有⼀个并⾏的reduceByKey操作可以返回⼀个分布数据集）。
在Spark所有的转化操作都是惰性求值的，就是说它们并不会⽴刻真的计算出结果。相反，它们仅仅是记录下了转换操作的操作对象（⽐如：⼀个⽂件）。只有当⼀个启动操作被执⾏，要向驱动程序返回结果时，转化操作才会真的开始计算。这样的设计使得Spark运⾏更加⾼效——⽐如，我们会发觉由map操作产⽣的数据集将会在reduce操作中⽤到，之后仅仅是返回了reduce的最终的结果⽽不是map产⽣的庞⼤数据集。
在默认情况下，每⼀个由转化操作得到的RDD都会在每次执⾏启动操作时重新计算⽣成。但是，你也可以通过调⽤persist(或cache)⽅法来将RDD持久化到内存中，这样Spark就可以在下次使⽤这个数据集时快速获得。Spark同样提供了对将RDD持久化到硬盘上或在多个节点间复制的⽀持。
下⾯的表格列出了Spark⽀持的常⽤转化操作。欲知细节，请查阅RDD API⽂档（, , ）和键值对RDD函数⽂档（, ）。
转化操作 | 作⽤
————| ——
map(func) | 返回⼀个新的分布数据集，由原数据集元素经func处理后的结果组成
filter(func) | 返回⼀个新的数据集，由传给func返回True的原数据集元素组成
flatMap(func) | 与map类似，但是每个传⼊元素可能有0或多个返回值，func可以返回⼀个序列⽽不是⼀个值
mapParitions(func) | 类似map，但是RDD的每个分⽚都会分开独⽴运⾏，所以func的参数和返回值必须都是迭代器
mapParitionsWithIndex(func) | 类似mapParitions，但是func有两个参数，第⼀个是分⽚的序号，第⼆个是迭代器。返回值还是迭代器
sample(withReplacement, fraction, seed) | 使⽤提供的随机数种⼦取样，然后替换或不替换
union(otherDataset) | 返回新的数据集，包括原数据集和参数数据集的所有元素
intersection(otherDataset) | 返回新数据集，是两个集的交集
distinct([numTasks]) | 返回新的集，包括原集中的不重复元素
groupByKey([numTasks]) | 当⽤于键值对RDD时返回(键，值迭代器)对的数据集
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) | ⽤于键值对RDD时返回（K，U）对集，对每⼀个Key的value进⾏聚集计算
sortByKey([ascending], [numTasks])⽤于键值对RDD时会返回RDD按键的顺序排序，升降序由第⼀个参数决定
join(otherDataset, [numTasks]) | ⽤于键值对(K, V)和(K, W)RDD时返回(K, (V, W))对RDD
cogroup(otherDataset, [numTasks]) | ⽤于两个键值对RDD时返回(K, (V迭代器， W迭代器))RDD
cartesian(otherDataset) | ⽤于T和U类型RDD时返回(T, U)对类型键值对RDD
pipe(command, [envVars]) | 通过shell命令管道处理每个RDD分⽚
coalesce(numPartitions) | 把RDD的分⽚数量降低到参数⼤⼩
repartition(numPartitions) | 重新打乱RDD中元素顺序并重新分⽚，数量由参数决定
repartitionAndSortWithinPartitions(partitioner) | 按照参数给定的分⽚器重新分⽚，同时每个分⽚内部按照键排序
具体⽰例：
map
将函数作⽤于数据集的每⼀个元素上，⽣成⼀个分布式的数据集返回
Return a new RDD by applying a function to each element of this RDD.
>>> rdd = sc.parallelize(["b", "a", "c"])
>>> sorted(rdd.map(lambda x: (x, 1)).collect())
[('a', 1), ('b', 1), ('c', 1)]
⼀个完整的例⼦：
from <span class='wp_keywordlink_affiliate'><a href="www.168seo/tag/pyspark" title="View all posts in pyspark" target="_blank">pyspark</a></span> import SparkConf,SparkContext
#配置
conf = SparkConf() #.setAppName("spark demo ").setMaster("local[2]")
sc = SparkContext(conf=conf)
data = range(10)
print(list(data))
r1 = sc.parallelize(data)
r2 = r1.map(lambda x:x+1)
llect())
sc.stop()
结果是：
filter
返回所有 funtion 返回值为True的函数，⽣成⼀个分布式的数据集返回
Return a new RDD containing only the elements that satisfy a predicate.
>>> rdd = sc.parallelize([1, 2, 3, 4, 5])
>>> rdd.filter(lambda x: x % 2 == 0).collect()
[2, 4]
⼀个完整的例⼦：
from pyspark import SparkConf,SparkContext
#配置
conf = SparkConf() #.setAppName("spark demo ").setMaster("local[2]")
sc = SparkContext(conf=conf)
data = range(10)
print(list(data))
r1 = sc.parallelize(data)
r2 = r1.filter(lambda x:x>5)
llect())
sc.stop()
结果是：
flatMap
Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.⼀个完整的例⼦：
from pyspark import SparkConf,SparkContext
#配置
conf = SparkConf() #.setAppName("spark demo ").setMaster("local[2]")
sc = SparkContext(conf=conf)
data = ["hello zeropython","hello 168seo"]
# print(list(data))
r1 = sc.parallelize(data)
r2 = r1.flatMap(lambda x:x.split(""))
r3 = r1.map(lambda x:x.split(""))
llect())
llect())
sc.stop()
RDD, and then flattening the results.
结果是：
groupBykey
按照相同key的数据分成⼀组
from _operator import add
from pyspark import SparkConf,SparkContext
#配置
conf = SparkConf() #.setAppName("spark demo ").setMaster("local[2]")
sc = SparkContext(conf=conf)
sc.setLogLevel("WARN")
"""
Return a new RDD by first applying a function to all elements of this
RDD, and then flattening the results.
"""
data = ["hello zeropython","hello 168seo"]
# print(list(data))
r1 = sc.parallelize(data)
r2 = r1.flatMap(lambda x:x.split("")).map(lambda y:(y,1))
print("r2",r2.collect())
r3 = r2.groupByKey()
print("r3",r3.collect())
r4 = r3.map(lambda x:{x[0]:list(x[1])})
print("r4",r4.collect())
duceByKey(add).collect())
sc.stop()
结果是：
groupBy运算
groupBy运算可以按照传⼊匿名函数的规则，将数据分为多个Array。⽐如下⾯的代码将intRDD分为偶数和奇数：result = upBy(lambda x : x % 2).collect()
print (sorted([(x, sorted(y)) for (x, y) in result]))
输出为：
[(0, [2]), (1, [1, 3, 5, 5])]
reduceBykey
把相同的key 的数据分发到⼀起并进⾏运算
from _operator import add
from pyspark import SparkConf,SparkContext
#配置
conf = SparkConf() #.setAppName("spark demo ").setMaster("local[2]")
sc = SparkContext(conf=conf)
data = ["hello zeropython","hello 168seo"]
# print(list(data))
r1 = sc.parallelize(data)
r2 = r1.flatMap(lambda x:x.split("")).map(lambda x:(x,1))
print("r2",r2.collect())
r3 = r2.reduceByKey(lambda x,y:x+y)
print("r3",r3.collect())
sc.stop()
结果是：
sortbykey
Sorts this RDD, which is assumed to consist of (key, value) pairs. from _operator import add
from pyspark import SparkConf,SparkContext
#配置
conf = SparkConf() #.setAppName("spark demo ").setMaster("local[2]")
sc = SparkContext(conf=conf)
# sc.setLogLevel("FATAL")
# sc.setLogLevel("ERROR")
sc.setLogLevel("ERROR")
data = ["hello zeropython","hwlldsf world","168seo","168seo","hello 168seo"] # print(list(data))
r1 = sc.parallelize(data)
r2 = r1.flatMap(lambda x:x.split(""))\
.map(lambda y:(y,1))\
.reduceByKey(lambda x,y:x+y)\
.sortByKey(lambda x:x[1])
# sortByKey排序根据关键词的值进⾏排序
# reduceByKey 让[("a",[1,1,1,1])] 转换成 [("a",3)]
llect())
sc.stop()
结果是：

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。