pyspark package

subpackages

pyspark.sql module
pyspark.streaming module
pyspark.ml package
pyspark.mllib package

PySpark是Spark的Python API。
Public classes:

SparkContext:
Spark功能的主要入口點(diǎn)。
RDD:
彈性分布式數(shù)據(jù)集（RDD），Spark中的基本抽象。
Broadcast:
一個(gè)廣播變量，可以在任務(wù)間重用。
Accumulator:
任務(wù)只能添加值的“add-only”共享變量。
SparkConf:
用于配置Spark。
SparkFiles:
訪問作業(yè)附帶的文件。
StorageLevel:
更細(xì)粒度的緩存持久化級(jí)別。
TaskContext:
有關(guān)當(dāng)前正在運(yùn)行的任務(wù)的信息，可在workers和實(shí)驗(yàn)室中獲得。

class pyspark.SpackConf(loadDefaults=True,_jvm=None,_jconf=None)

Spark應(yīng)用程序的配置。用于設(shè)置spark變量，參數(shù)以鍵值對的形式傳遞。大多數(shù)情況下，您將使用SparkConf（）創(chuàng)建一個(gè)SparkConf對象，該對象將從spark.* Java系統(tǒng)屬性中加載值。在這種情況下，您直接在SparkConf對象上設(shè)置的任何參數(shù)都優(yōu)先于系統(tǒng)屬性。

對于單元測試，您也可以調(diào)用SparkConf（false）來跳過加載外部設(shè)置并獲取相同的配置，而不管系統(tǒng)屬性是什么。這個(gè)類中的所有setter方法都支持鏈接。例如，你可以寫conf.setMaster("local").setAppName("Myapp")。

注意:一旦SparkConf對象被傳遞給Spark，它就被克隆，不能再被用戶修改。

contains(key)
這個(gè)配置是否包含給定的key？
get(key, defaultValue=None)
獲取配置中某個(gè)key的value，否則返回默認(rèn)值。
getAll()
獲取所有的值，以key-value pairs list的形式。
set(key, value)
設(shè)置配置屬性。
setAll(pairs)
設(shè)置多個(gè)參數(shù)，以key-value pairs list的形式傳遞。
setAppName(value)
設(shè)置application的name。
setExecutorEnv(key=None, value=None, pairs=None)
設(shè)置要傳遞給executors的環(huán)境變量。
setIfMissing(key,value)
設(shè)置配置屬性（如果尚未設(shè)置）。
setMaster(value)
設(shè)置要連接到的 master URL。
setSparkHome(value)
設(shè)置在worker nodes安裝spark的path
toDebugString()
以key=value pairs list的形式返回配置的可打印版本，每行一個(gè)。

class 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'>)

Spark功能的主要入口點(diǎn)。 SparkContext表示與Spark集群的連接，可用于在該集群上創(chuàng)建RDD和廣播變量。
PACKAGE_EXTENSIONS = ('.zip', '.egg', '.jar')

accumulator(value, accum_param=None)
用給定的初始值創(chuàng)建一個(gè)累加器(accumulator)，如果提供的話，使用給定的AccumulatorParam 幫助對象來定義如何 add 某種數(shù)據(jù)類型的值。默認(rèn)AccumulatorParams用于整數(shù)和浮點(diǎn)數(shù)，如果你沒有提供。對于其他類型，可以使用自定義的AccumulatorParam。
addFile(path, recursive=False)
在每個(gè)節(jié)點(diǎn)上添加一個(gè)文件，用這個(gè)Spark job下載。傳遞的路徑可以是本地文件，HDFS中的文件（或其他Hadoop支持的文件系統(tǒng)），也可以是HTTP，HTTPS或FTP URI。要訪問Spark jobs中的文件，請使用文件名L {SparkFiles.get（fileName）<pyspark.files.SparkFiles.get>}來查找其下載位置。如果遞歸(recursive)選項(xiàng)設(shè)置為True，則可以給出一個(gè)目錄。當(dāng)前目錄僅支持Hadoop支持的文件系統(tǒng)。

#下面代碼通過鍵入 bin/pysaprk 命令進(jìn)入交互式模式運(yùn)行，sc為sparkContext()的實(shí)例對象,spark已經(jīng)為我們自動(dòng)創(chuàng)建好了。
import os
from pyspark import SparkFiles
with open('/home/test.txt', 'w') as f: #在本地建一個(gè)文件并寫入內(nèi)容
    f.write('100')
#加載文件，若是hdfs中的文件可以這樣寫：
# hdfs://hostname:port/file/path
# 例如 hdfs://master:9000/user/hadoop/test.txt
sc.addFile('/home/test.txt') 
def func(iterator):
    with open(SparkFiles.get('test.txt')) as testFile:
        fileVal = int(testFile.readline())
        return [x * fileVal for x in iterator]
sc.parallelize([1,2,3,4]).mapPartitions(func).collect() #創(chuàng)建一個(gè)RDD并把函數(shù)func作用到每個(gè)元素上，最后輸出結(jié)果。
[100, 200, 300, 400] #output

addPyFile(path)
為將來在此SparkContext上執(zhí)行的所有tasks添加一個(gè).py或.zip依賴項(xiàng)。傳遞的路徑(path)可以是本地文件，HDFS中的文件（或Hadoop支持的其他文件系統(tǒng)），也可以是HTTP，HTTPS或FTP URI。
applicationId
Spark應(yīng)用程序的唯一標(biāo)識(shí)符。其格式取決于調(diào)度程序的實(shí)現(xiàn)。

例如 local spark app something like 'local-1433865536131'
in case of YARN something like ‘a(chǎn)pplication_1433865536131_34483’

sc.applicationId
u'local-1513691270280' #output

binaryFiles(path, minPartitions=None)
從HDFS，本地文件系統(tǒng)（所有節(jié)點(diǎn)上都可用）或Hadoop支持的任何文件系統(tǒng)URI讀取二進(jìn)制文件的目錄作為字節(jié)數(shù)組。每個(gè)被讀取的文件作為單個(gè)記錄，并返回一個(gè)key-value pair，其中key是每個(gè)文件的路徑，value是每個(gè)文件的內(nèi)容。

Note:小文件是首選，大文件也是允許的，但可能會(huì)導(dǎo)致性能不佳。

binaryRecords(path, recordLength)

Note: 實(shí)驗(yàn)階段(Experimental)
從一個(gè)扁平的二進(jìn)制文件加載數(shù)據(jù)，假定每個(gè)記錄都是一組具有指定數(shù)字格式的數(shù)字（請參閱ByteBuffer），并且每個(gè)記錄的字節(jié)數(shù)是恒定的。
參數(shù)(Parmenters):
path - 輸入數(shù)據(jù)文件的目錄;
recordLength - 記錄的分割長度

broadcast(value)
將一個(gè)只讀變量廣播到集群，返回一個(gè)L {Broadcast <pyspark.broadcast.Broadcast>}對象來在分布式函數(shù)中讀取它。變量只會(huì)發(fā)送到每個(gè)群集一次。
cancelAllJobs()
取消所有已被調(diào)度的或正在運(yùn)行的作業(yè)。
cancelJobGroup(groupId)
取消指定組的活動(dòng)作業(yè)。有關(guān)更多信息，請參見SparkContext.setJobGroup。
defaultMinPartitions
當(dāng)用戶不給定時(shí)，Hadoop RDD分區(qū)的默認(rèn)最小數(shù)量。
defaultParallelism
當(dāng)用戶沒有給出時(shí)，默認(rèn)的并行度級(jí)別（例如 reduce tasks）
dump_profiles(path)
將配置文件轉(zhuǎn)儲(chǔ)到目錄路徑
emptyRDD()
創(chuàng)建一個(gè)沒有分區(qū)或元素的RDD。
getConf()
getLocalProperty(key)
獲取在此線程中設(shè)置的本地屬性，如果缺失，則返回null。請參閱setLocalProperty
classmethod getOrCreate(conf=None)
獲取或?qū)嵗粋€(gè)SparkContext并將其注冊為一個(gè)單例對象。
parameters:
conf - SparkConf(optional)
hadoopFile(path, inputFormatClass, keyClass, valueClass, keyConverter=None, valueConverter=None, conf=None,batchSize=0)
使用HDFS，本地文件系統(tǒng)（所有節(jié)點(diǎn)上都可用）或任何Hadoop支持的文件系統(tǒng)URI讀取具有任意鍵和值類的“old”Hadoop InputFormat。機(jī)制與sc.sequenceFile相同。Hadoop配置可以作為Python字典的形式傳入。這將被轉(zhuǎn)換成Java中的配置。
parameters:
path - hadoop 文件的路徑；
inputFormatClass - Hadoop InputFormat的完全限定類名(e.g. "org.apache.hadoop.mapred.TextInputFormat")；
keyClass - key Writable類的完全限定類名(e.g. "org.apache.hadoop.io.Text")；
valueClass - value Writable類的完全限定類名(e.g. "org.apache.hadoop.io.LongWritable")；
keyConverter - (None by default)；
valueConverter - (None by default)；
conf - 配置Hadoop，以dict的形式傳入(None by default)；
batchSize - 表示為單個(gè)Java對象的Python對象的數(shù)量。(默認(rèn)為0，batchSize自動(dòng)選擇)
hadoopRDD(inputFormatClass, keyClass, valueClass, keyConverter=None, valueConverter=None, conf=None, batchSize=0)
從任意的Hadoop配置中讀取具有任意鍵和值類的“old”Hadoop InputFormat，以Python dict 的形式傳入。這將被轉(zhuǎn)換成Java中的配置。機(jī)制與sc.sequenceFile相同。
newAPIHadoopFile(path, inputFormatClass, keyClass, valueClass, keyConverter=None, valueConverter=None, conf=None, batchSize=0)
使用HDFS，本地文件系統(tǒng)（所有節(jié)點(diǎn)上都可用）或任何Hadoop支持的文件系統(tǒng)URI讀取具有任意鍵和值類的“new API”Hadoop InputFormat。機(jī)制與sc.sequenceFile相同。hadoop 配置可以以python中dict的形式被傳入，這將被轉(zhuǎn)換成Java中的配置。
newAPIHadoopRDD(inputFormatClass, keyClass, valueClass, keyConverter=None, valueConverter=None, conf=None, batchSize=0)
從任意的Hadoop配置中讀取具有任意鍵和值類的“new API”Hadoop InputFormat，以Python dict 的形式傳入。這將被轉(zhuǎn)換成Java中的配置。機(jī)制與sc.sequenceFile相同。
parallelize(c, numSlices=None)
分發(fā)本地Python集合轉(zhuǎn)成RDD形式。如果輸入表示性能范圍，則建議使用xrange。

sc.parallelize([0, 2, 3, 4, 6], 5) #分成5個(gè)partitions
[[0], [2], [3], [4], [6]] # output
sc.parallelize(xrange(0, 6, 2),5)
[[], [0], [], [2], [4]] # output

pickleFile(name, minPartitions=None)
加載之前使用RDD.saveAsPickleFile方法保存的RDD。

from tempfile import NamedTemporaryFile
#創(chuàng)建一個(gè)有名字的臨時(shí)文件,并自動(dòng)刪除
tmpFile = NamedTemporaryFile(delete=True) 
tmpFile.close() #關(guān)閉這個(gè)臨時(shí)文件
sc.parallelize(range(10)).saveAsPickleFile(tmpFile.name, 5)
sorted(sc.pickleFile(tmpFile.name, 3).collect())
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9] #output

range(start, end=None, step=1, numSlices=None)
創(chuàng)建一個(gè)新的包含從開始到結(jié)束（獨(dú)占）的元素的int類型的RDD，通過step增加每個(gè)元素。可以像python內(nèi)置的range（）函數(shù)一樣調(diào)用。如果使用單個(gè)參數(shù)調(diào)用，參數(shù)被解釋為結(jié)束，并且start被設(shè)置為0。numSlices - 新RDD的分區(qū)數(shù)量，返回Int類型的RDD。

sc.range(5).collect()
[0, 1, 2, 3, 4]
sc.range(2, 4).collect()
[2, 3]
sc.range(1, 7, 2).collect()
[1, 3, 5]

runJob(rdd, partitionFunc, partitions=None, allowLocal=False)
在指定的一組分區(qū)上執(zhí)行給定的partitionFunc，并將結(jié)果作為元素?cái)?shù)組的形式返回。如果未指定“partitions”，則將在所有分區(qū)上運(yùn)行。

myRDD = sc.parallelize(range(6), 3)
sc.runJob(myRDD, lambda part: [x * x for x in part])
[0, 1, 4, 9, 16, 25]
myRDD =  sc.parallelize(range(6), 3)
myRDD.glom().collect()
[[0, 1], [2, 3], [4, 5]]
#匿名函數(shù)只作用在0和2兩個(gè)分區(qū)上
sc.runJob(myRDD, lambda part: [x * x for x in part],[0,2], True) 
[0, 1, 16, 25]

sequenceFile(path, keyClass=None, valueClass=None, keyConverter=None, valueConverter=None, minSplits=None, batchSize=0)
從HDFS，本地文件系統(tǒng)（所有節(jié)點(diǎn)都可用）或任何Hadoop支持的文件系統(tǒng)URI讀取任意鍵和值 Writable class 的Hadoop SequenceFile。minSplits - 數(shù)據(jù)集中的最小分割(default min(2, sc.defaultParallelism))
setCheckpointDir(dirName)
設(shè)置將RDD作為檢查點(diǎn)的目錄。如果在群集上運(yùn)行，該目錄必須是HDFS路徑。
setJobGroup(groupId, description, interruptOnCancel=False)
為此線程啟動(dòng)的所有作業(yè)分配一個(gè)組ID，直到組ID被設(shè)置為不同的值或清除。通常，應(yīng)用程序中的執(zhí)行單元由多個(gè)Spark actions或jobs組成。應(yīng)用程序員可以使用這種方法將所有這些作業(yè)分組在一起，并給出一個(gè)組描述。一旦設(shè)置，Spark Web UI將把這些作業(yè)與這個(gè)組關(guān)聯(lián)起來。應(yīng)用程序可以使用SparkContext.cancelJobGroup取消該組中的所有正在運(yùn)行的jobs。

import threading
from time import sleep
result = "Not Set"
lock = threading.Lock() #定義一個(gè)鎖
def map_func(x):
    sleep(100)
    raise Exception("Task should have been cancelled")
def start_job(x):
    global result
    try:
        sc.setJobGroup("job_to_cancel", "some description")
        result = sc.parallelize(range(x)).map(map_func).collect()
    except Exception as e:
        result = "Cancelled"
    lock.release() #打開鎖，釋放資源
def stop_job():
    sleep(5)
    sc.cancelJobGroup("job_to_cancel")
supress = lock.acquire() #鎖定資源
supress = threading.Thread(target=start_job, args=(10, )).start()
supress = threading.Thread(target=stop_job).start()
supress = lock.acquire()
print(result)
Cancelled #output

如果作業(yè)組的interruptOnCancel設(shè)置為true，那么作業(yè)取消將導(dǎo)致在作業(yè)的執(zhí)行程序線程上調(diào)用Thread.interrupt（）。這有助于確保任務(wù)實(shí)際上被及時(shí)停止，但由于HDFS-1208的原因，默認(rèn)情況下會(huì)關(guān)閉，HDFS可能會(huì)通過將節(jié)點(diǎn)標(biāo)記為死亡來響應(yīng)Thread.interrupt（）。

setLocalProperty(key, value)
設(shè)置一個(gè)本地屬性，將影響從此線程提交的作業(yè)，例如Spark Fair Scheduler pool。
setLogLevel(logLevel)
控制我們的logLevel。這將覆蓋任何用戶定義的日志設(shè)置。有效的日志級(jí)別包括：ALL，DEBUG，ERROR，F(xiàn)ATAL，INFO，OFF，TRACE，WARN
classmethod setSystemProperty(key, value)
設(shè)置一個(gè)Java系統(tǒng)屬性，如spark.executor.memory。這必須在實(shí)例化SparkContext之前調(diào)用。
show_profiles()
將配置文件統(tǒng)計(jì)信息打印到標(biāo)準(zhǔn)輸出
sparkUser()
為用戶獲取SPARK_USER，運(yùn)行SparkContext的用戶。
startTime
返回Spark Context啟動(dòng)時(shí)的紀(jì)元時(shí)間。
statusTracker()
返回StatusTracker對象
stop()
終止SparkContext
textFile(name, minPartitions=None, use_unicode=True)
從HDFS、本地文件系統(tǒng)（所有節(jié)點(diǎn)都可用）或任何Hadoop支持的文件系統(tǒng)URI讀取文本文件，并將其作為字符串的RDD返回。如果use_unicode為False，則字符串將保持為str（編碼為utf-8），這比unicode更快更小。(在Spark 1.2中添加)

with open('/home/test.txt', 'w') as testFile:
    testFile.write("Hello world!")
#讀取本地文件。例子： file:/home/test.txt
testFile = sc.textFile('file:/home/test.txt')
testFile.collect()
[u'Hello world!'] #output

uiWebUrl
返回由此SparkContext啟動(dòng)的SparkUI實(shí)例的URL
union(rdds)
建立一個(gè)RDD列表的聯(lián)合。這支持具有不同序列化格式的RDDs的unions()，盡管這迫使它們使用默認(rèn)序列化器被重新序列化。

path = os.path.join('/home/', 'union-text.txt')
with open(path, 'w') as testFile:
    testFile.write("hello")
textFile = sc.textFile(path)
textFile.collect()
[u'Hello'] #output
parallelize = sc.parallelize(["World!"])
#union類似于列表的append方法
sorted(sc.union([textFile, parallelize]).collect())
[u'Hello', 'World!'] #output

version
運(yùn)行此應(yīng)用程序的Spark版本。
wholeTextFiles(path, minPartitions=None, use_unicode=True)
從HDFS，本地文件系統(tǒng)（所有節(jié)點(diǎn)都可用）或任何支持Hadoop的文件系統(tǒng)URI讀取文本文件的目錄。每個(gè)文件被讀取為單個(gè)記錄，并返回一個(gè)鍵值對，其中鍵是每個(gè)文件的路徑，值是每個(gè)文件的內(nèi)容。如果use_unicode為False，則字符串將保持為str（編碼為utf-8），這比unicode更快更小，(在Spark 1.2中添加)。小文件是首選，因?yàn)槊總€(gè)文件將被完全加載到內(nèi)存中。

dirPath = os.path.join(/home/, 'files')
os.mkdir(dirPath)
with open(os.path.join(dirPath, '1.txt'), 'w') as file1:
    file1.write('1')
with open(os.path.join(dirPath, '2.txt'), 'w') as file2:
    file2.write('2')
textFiles = sc.wholeTextFiles(dirPath)
sorted(textFiles.collect())
[(u'.../1.txt', u'1'), (u'.../2.txt', u'2')]

class pyspark.SparkFiles

解析通過L {SparkContext.addFile（）<pyspark.context.SparkContext.addFile>}添加的文件的路徑。parkFiles僅包含類方法; 用戶不應(yīng)該創(chuàng)建SparkFiles實(shí)例。
classmethod get(filename)
獲取通過SparkContext.addFile（）添加的文件的絕對路徑。
classmethod getRootDirectory()
獲取通過SparkContext.addFile（）添加的文件的根目錄。

class pyspark.RDD(jrdd, ctx, jrdd_deserializer=AutoBatchedSerializer(PickleSerializer()))

彈性分布式數(shù)據(jù)集（RDD），Spark中的基本抽象。表示可以并行操作的不可變分區(qū)元素集合。

aggregate(zeroValue, seqOp, combOp)
先聚合每個(gè)分區(qū)中的元素，然后使用給定的組合函數(shù)和一個(gè)中性“零值”對所有分區(qū)的結(jié)果進(jìn)行聚合。
函數(shù)op（t1，t2）允許修改t1并將其返回為結(jié)果值，以避免對象分配; 但是，該函數(shù)不應(yīng)該修改t2。
第一個(gè)函數(shù)（seqOp）可以返回與此RDD類型不同的結(jié)果類型U. 因此，我們需要一個(gè)合并T到U的操作和一個(gè)合并兩個(gè)U的操作。

seqOp = (lambda x, y: (x[0] + y, x[1] + 1))
combOp = (lambda x, y: (x[0] + y[0], x[1] + y[1]))
sc.parallelize([1, 2, 3, 4]).aggregate((0, 0), seqOp, combOp)
(10, 4) #output
sc.parallelize([]).aggregate((0, 0), seqOp, combOp)
(0, 0) #output
解釋如下：
假設(shè)[1,2,3,4]被分成兩個(gè)分區(qū)，為 分區(qū)1（[1,2]），分區(qū)2（[3,4]）
seqOp作用在每一個(gè)分區(qū)上
首先用seqOp對分區(qū)1進(jìn)行操作：
       x=(0,0)   y=1    -->   (1,1)   #對分區(qū)進(jìn)行第一次seqOp操作時(shí)，x為zero value
       x=(1,1)   y=2    -->   (3,2)   #對分區(qū)進(jìn)行的第二次及以后的seqOp操作，x為前一次seqOp的執(zhí)行結(jié)果
       同樣對分區(qū)2進(jìn)行操作：
       x=(0,0)   y=3    -->   (3,1) 
       x=(3,1)   y=4    -->   (7,2)
然后用combOp對兩個(gè)分區(qū)seqOp作用后的結(jié)果進(jìn)行操作：
      分區(qū)1：
      x=(0,0)   y=(3,2)   > (3,2)  #對第一個(gè)分區(qū)進(jìn)行combOp操作時(shí)，x為zero value
      分區(qū)2：
      x=(3,2)   y=(7,2)   > (10,4) #對第二個(gè)及以后分區(qū)進(jìn)行combOp操作時(shí)，x為前一分區(qū)combOp處理后的結(jié)果

aggregateByKey(zeroValue, seqFunc, combFunc, numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)
使用給定的combine(組合)函數(shù)和一個(gè)中性的“零值”來聚合每個(gè)鍵的值。這個(gè)函數(shù)可以返回一個(gè)不同的結(jié)果類型U，比這個(gè)RDD中的值的類型V.因此，我們需要一個(gè)操作來合并一個(gè)V到一個(gè)U和一個(gè)合并兩個(gè)U的操作，前一個(gè)操作用于合并分區(qū)內(nèi)的值，后者用于合并分區(qū)之間的值。為了避免內(nèi)存分配，這兩個(gè)函數(shù)都允許修改并返回它們的第一個(gè)參數(shù)，而不是創(chuàng)建一個(gè)新的U.
cache()
使用默認(rèn)存儲(chǔ)級(jí)別（MEMORY_ONLY）持久化RDD。
cartesian(other)
返回一個(gè)RDD和另一個(gè)RDD的笛卡爾乘積，即所有元素對（a，b）的RDD，其中a在自身中，b在另一個(gè)中。

rdd = sc.parallelize([1, 2])
sorted(rdd.cartesian(rdd).collect())
[(1, 1), (1, 2), (2, 1), (2, 2)] #output

checkpoint()
將此RDD標(biāo)記為檢查點(diǎn)。它將被保存到使用SparkContext.setCheckpointDir（）設(shè)置的檢查點(diǎn)目錄內(nèi)的文件中，并且對其父RDD的所有引用都將被刪除。在此RDD上執(zhí)行任何作業(yè)之前，必須調(diào)用此函數(shù)。強(qiáng)烈建議將此RDD保存在內(nèi)存中，否則將其保存在文件中將需要重新計(jì)算。
coalesce(numPartitions, shuffle=False)
返回一個(gè)減少到numPartitions分區(qū)的新RDD。
把RDD的分區(qū)數(shù)降低到通過參數(shù)numPartitions指定的值。在得到的更大一些數(shù)據(jù)集上執(zhí)行操作，會(huì)更加高效。

sc.parallelize([1, 2, 3, 4, 5], 3).glom().collect()
[[1], [2, 3], [4, 5]]
sc.parallelize([1, 2, 3, 4, 5], 3).coalesce(1).glom().collect()
[[1, 2, 3, 4, 5]]

cogroup(other, numPartitions=None)
對于自己或其他中的每個(gè)關(guān)鍵字k，返回一個(gè)包含一個(gè)元組的結(jié)果RDD，以及該關(guān)鍵字在自身以及其他值中的值列表。

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2)])
[(x, tuple(map(list, y))) for x, y in sorted(list(x.cogroup(y).collect()))]
[('a', ([1], [2])), ('b', ([4], []))]

collect()
返回包含此RDD中所有元素的列表。
在驅(qū)動(dòng)程序中，以數(shù)組的形式返回?cái)?shù)據(jù)集的所有元素。通常用于filter或其它產(chǎn)生了大量小數(shù)據(jù)集的情況。

Note:因?yàn)樗械臄?shù)據(jù)都被加載到驅(qū)動(dòng)程序的內(nèi)存中，所以只能在結(jié)果數(shù)組很小的情況下使用此方法。

collectAsMap()
將此RDD中的鍵值對作為字典返回給master。

Note: 因?yàn)樗械臄?shù)據(jù)都被加載到驅(qū)動(dòng)程序的內(nèi)存中，所以只有在結(jié)果數(shù)據(jù)很小的情況下才能使用此方法。

m = sc.parallelize([(1, 2), (3, 4)]).collectAsMap()
print(m)
{1: 2, 3: 4}
m[1]
2

combineByKey(createCombiner, mergeValue, mergeCombiners, numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)
通用函數(shù)使用一組自定義的聚合函數(shù)來組合每個(gè)鍵的元素。
將RDD [（K，V）]轉(zhuǎn)換為類型RDD [（K，C）]的結(jié)果，以用于“combined type”C.
使用者提供三個(gè)函數(shù)：

createCombiner，which turns a V into a C (e.g., creates a one-element list)
mergeValue，to merge a V into a C (e.g., adds it to the end of a list)
mergeCombiners，to combine two C’s into a single one (e.g., merges the lists)
為了避免內(nèi)存分配，mergeValue和mergeCombiners都允許修改并返回它們的第一個(gè)參數(shù)，而不是創(chuàng)建一個(gè)新的C.
另外，用戶可以控制輸出RDD的分區(qū)。

x = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
def to_list(a):
  return [a]
def append(a, b):
  a.append(b)
  return a
def extend(a, b):
  a.extend(b)
  return a
sorted(x.combineByKey(to_list, append, extend).collect())
[('a', [1, 2]), ('b', [1])]
解釋如下：
由于聚合操作會(huì)遍歷分區(qū)中所有的元素，
因此每個(gè)元素（這里指的是鍵值對）的鍵只有兩種情況：出現(xiàn)過和沒出現(xiàn)過。
如果以前沒出現(xiàn)過，則執(zhí)行的是createCombiner函數(shù)，
在該例子中首先是('a',1)中key:a沒有出現(xiàn)過，
我們執(zhí)行createConmbiner函數(shù)作用在value:1上，變成了列表[1]。('b',1)同理。
如果key以前出現(xiàn)過就執(zhí)行mergeValue函數(shù)作用在value上，
('a',2)，由于key:a以前出現(xiàn)過所以就把value:2 append到列表[1]上，結(jié)果為[1,2]。
mergeCombiners函數(shù)是作用在每個(gè)分區(qū)上，因?yàn)橹挥幸粋€(gè)分區(qū)，所以并沒有用上這個(gè)函數(shù)。

context
此RDD創(chuàng)建的SparkContext。
count()
返回此RDD中的元素?cái)?shù)量。

sc.parallelize([2, 3, 4]).count()
3

countApprox(timeout, confidence=0.95)

Note: 實(shí)驗(yàn)階段
count（）的近似版本，即使并非所有任務(wù)都已完成，在超時(shí)內(nèi)返回潛在的不完整結(jié)果。

countApproxDistinct(relativeSD=0.05)

Note: 實(shí)驗(yàn)階段
返回RDD中不同元素的近似數(shù)量。

countByKey()
計(jì)算每個(gè)鍵的元素?cái)?shù)量，并將結(jié)果作為字典返回給master。

rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
sorted(rdd.countByKey().items())
[('a', 2), ('b', 1)]

countByValue()
將此RDD中每個(gè)唯一值的計(jì)數(shù)返回為（值，計(jì)數(shù)）對的字典。
sorted(sc.parallelize([1, 2, 1, 2, 2], 2).countByValue().items())
[(1, 2), (2, 3)]
distinct(numPartitions=None)
返回包含此RDD中不同元素的新RDD。去重。

sorted(sc.parallelize([1, 1, 2, 3]).distinct().collect())
[1,2,3]

filter(f)
返回僅包含滿足f的元素的新RDD。

rdd = sc.parallelize([1, 2, 3, 4, 5])
rdd.filter(lambda x: x % 2 == 0).collect()
[2,4]

first()
返回此RDD中的第一個(gè)元素。

sc.parallelize([2, 3, 4]).first()
2

flatMap(f, preservesPartitioning=False)
通過首先將一個(gè)函數(shù)應(yīng)用于此RDD的所有元素，然后展平結(jié)果，返回一個(gè)新的RDD。

rdd = sc.parallelize([2, 3, 4])
rdd.flatMap(lambda x: range(1, x)).collect()
[1, 1, 2, 1, 2, 3]
sc.parallelize([(1,2),(3,4)]).flatMap(lambda x: x).collect()
[1, 2, 3, 4]
sc.parallelize([(1,2),(3,4)]).map(lambda x: x).collect()
[(1, 2), (3, 4)]

flatMapValues(f)
通過flatMap函數(shù)傳遞鍵值對RDD中的每個(gè)值而不更改鍵; 這也保留了原來的RDD的分區(qū)。

x = sc.parallelize([("a", ["x", "y", "z"]), ("b", ["p", "r"])])
def f(x): 
  return x
x.flatMapValues(f).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

fold(zeroValue, op)
使用給定的關(guān)聯(lián)函數(shù)和中性“零值”來聚合每個(gè)分區(qū)的元素，然后聚合所有分區(qū)的結(jié)果。
函數(shù)op（t1，t2）允許修改t1并將其作為結(jié)果值返回，以避免對象分配; 但是，它不應(yīng)該修改t2。

from operator import add
sc.parallelize([1, 2, 3, 4, 5]).fold(0, add)
15

foldByKey(zeroValue, func, numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)

rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
from operator import add
sorted(rdd.foldByKey(0, add).collect())
[('a', 2), ('b', 1)]

foreach(f)
將一個(gè)函數(shù)應(yīng)用于此RDD的所有元素。

def f(x): 
  print(x)
sc.parallelize([1, 2, 3]).foreach(f)
3
2
1

foreachPartition(f)
將此功能應(yīng)用于此RDD的每個(gè)分區(qū)。
fullOuterJoin(other, numPartitions=None)
執(zhí)行self和other的右外連接。

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("c", 8)])
sorted(x.fullOuterJoin(y).collect())
[('a', (1, 2)), ('b', (4, None)), ('c', (None, 8))]

getCheckpointFile()
獲取此RDD檢查點(diǎn)所在的文件的名稱
RDD在本地檢查點(diǎn)不定義。
getNumPartitions()
返回RDD中的分區(qū)數(shù)量
getStorageLevel()
獲取RDD的當(dāng)前存儲(chǔ)級(jí)別。
glom()
將通過合并每個(gè)分區(qū)內(nèi)的所有元素創(chuàng)建的RDD返回到列表中。

rdd = sc.parallelize([1, 2, 3, 4], 2)
sorted(rdd.glom().collect())
[[1, 2], [3, 4]]

groupBy(f, numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)
通過函數(shù)f對RDD進(jìn)行分組。

rdd = sc.parallelize([1, 1, 2, 3, 5, 8])
result = rdd.groupBy(lambda x: x % 2).collect()
sorted([(x, sorted(y)) for (x, y) in result])
[(0, [2, 8]), (1, [1, 1, 3, 5])]

groupByKey(numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)
groupWith(other, *others)
cogroup的別名，但支持多個(gè)RDD。
histogram(buckets)
使用提供的桶計(jì)算直方圖。除了最后一個(gè)關(guān)閉之外，桶都向右開放。
桶必須被排序，不包含任何重復(fù)，并且至少有兩個(gè)元素。

rdd = sc.parallelize(range(51))
rdd.histogram(2) #兩個(gè)桶
([0, 25, 50], [25, 26]) 
#第一個(gè)桶在區(qū)間[1,25),包含25個(gè)元素，第二個(gè)桶在區(qū)間[25,50],包含26個(gè)元素。

id()
此RDD的唯一ID（在其SparkContext中）。
intersection(other)
返回此RDD與另一個(gè)的交集。

rdd1 = sc.parallelize([1, 10, 2, 3, 4, 5])
rdd2 = sc.parallelize([1, 6, 2, 3, 7, 8])
rdd1.intersection(rdd2).collect()
[1, 2, 3]

isCheckpointed()
返回這個(gè)RDD是否是檢查點(diǎn)和物化的，可靠的或本地的。
isEmpty()
當(dāng)且僅當(dāng)RDD完全不包含任何元素時(shí)才返回true。
isLocallyCheckpointed()
返回此RDD是否標(biāo)記為本地檢查點(diǎn)。
join(other, numPartitions=None)
返回一個(gè)RDD，其中包含所有自身和其他匹配鍵的元素對。
用于操作兩個(gè)鍵值對格式的數(shù)據(jù)集，操作兩個(gè)數(shù)據(jù)集(K,V)和(K,W)返回(K, (V, W))格式的數(shù)據(jù)集。通過leftOuterJoin、rightOuterJoin、fullOuterJoin完成外連接操作。
每一對元素將作為（k，（v1，v2））元組返回，其中（k，v1）是自身的，（k，v2）是另一個(gè)的。

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("a", 3)])
sorted(x.join(y).collect())
[('a', (1, 2)), ('a', (1, 3))]

keyBy(f)
通過應(yīng)用函數(shù)f來創(chuàng)建此RDD中元素的元組。

x = sc.parallelize(range(0,3)).keyBy(lambda x: x*x)
x.collect()
[(0, 0), (1, 1), (4, 2)]

keys()
每個(gè)元組的鍵返回一個(gè)RDD。

m = sc.parallelize([(1, 2), (3, 4)]).keys()
m.collect()
[1, 3]

leftOuterJoin(other, numPartitions=None)

x = sc.parallelize([("a", 1), ("b", 4)])
y = sc.parallelize([("a", 2), ("c", 8)])
x.fullOuterJoin(y).collect()
[('a', (1, 2)), ('c', (None, 8)), ('b', (4, None))]
x.leftOuterJoin(y).collect()
[('a', (1, 2)), ('b', (4, None))]
x.rightOuterJoin(y).collect()
[('a', (1, 2)), ('c', (None, 8))]

localCheckpoint()
使用Spark的現(xiàn)有緩存層將此RDD標(biāo)記為本地檢查點(diǎn)。
lookup(key)
返回RDD中鍵值的列表。如果RDD具有已知的分區(qū)程序，則只需搜索鍵映射到的分區(qū)即可高效地執(zhí)行此操作。

 sc.parallelize([('a',1),('b',2)]).lookup('b')
[2]

map(f, preservesPartitioning=False)
通過對這個(gè)RDD的每個(gè)元素應(yīng)用一個(gè)函數(shù)來返回一個(gè)新的RDD。

rdd = sc.parallelize(["b", "a", "c"])
sorted(rdd.map(lambda x: (x, 1)).collect())
[('a', 1), ('b', 1), ('c', 1)]

mapPartitions(f, preservesPartitioning=False)
通過對該RDD的每個(gè)分區(qū)應(yīng)用一個(gè)函數(shù)來返回一個(gè)新的RDD。

rdd = sc.parallelize([1, 2, 3, 4], 2)
rdd.glom().collect()
[[1, 2], [3, 4]]
def f(iterator): 
  yield sum(iterator)
rdd.mapPartitions(f).collect()
[3, 7]

mapPartitionsWithIndex(f, preservesPartitioning=False)
通過對該RDD的每個(gè)分區(qū)應(yīng)用函數(shù)f，同時(shí)跟蹤原始分區(qū)的索引，返回新的RDD。
mapPartitionsWithSplit(f, preservesPartitioning=False)

棄用，用mapPartitionsWithIndex(f, preservesPartitioning=False)代替

mapValues(f)
通過映射函數(shù)傳遞鍵值對RDD中的每個(gè)值，而不更改鍵; 這也保留了原來的RDD的分區(qū)。

x = sc.parallelize([("a", ["apple", "banana", "lemon"]), ("b", ["grapes"])])
def f(x): return len(x)
x.mapValues(f).collect()
[('a', 3), ('b', 1)]

max(key=None)
找到這個(gè)RDD中的最大項(xiàng)。
mean()
計(jì)算這個(gè)RDD元素的平均值。
meanApprox(timeout, confidence=0.95)

實(shí)驗(yàn)階段

min(key=None)
找到此RDD中的最小項(xiàng)。
name()
返回此RDD的名稱。
partitionBy(numPartitions, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)
使用指定的分區(qū)程序返回分區(qū)的RDD副本。
persist(storageLevel=StorageLevel(False, True, False, False, 1))
設(shè)置此RDD的存儲(chǔ)級(jí)別以便在第一次計(jì)算之后在操作之間保留其值。如果RDD尚未設(shè)置存儲(chǔ)級(jí)別，則只能用于分配新的存儲(chǔ)級(jí)別。如果未指定存儲(chǔ)級(jí)別，則默認(rèn)為（MEMORY_ONLY）。
pipe(command, env=None, checkCode=False)
以管道(pipe)方式將 RDD的各個(gè)分區(qū)(partition)使用 shell命令處理（比如一個(gè) Perl或 bash腳本）。 RDD的元素會(huì)被寫入進(jìn)程的標(biāo)準(zhǔn)輸入(stdin)，將進(jìn)程返回的一個(gè)字符串型 RDD(RDD of strings)，以一行文本的形式寫入進(jìn)程的標(biāo)準(zhǔn)輸出(stdout)中。
randomSplit(weights, seed=None)
根據(jù)提供的權(quán)重隨機(jī)分配這個(gè)RDD。

rdd = sc.parallelize(range(500), 1)
rdd1, rdd2 = rdd.randomSplit([2, 3], 17)
rdd1.count()
213
rdd2.count()
287

reduce(f)
使用函數(shù)func聚集數(shù)據(jù)集中的元素，這個(gè)函數(shù)func輸入為兩個(gè)元素，返回為一個(gè)元素。這個(gè)函數(shù)應(yīng)該符合結(jié)合律和交換律，這樣才能保證數(shù)據(jù)集中各個(gè)元素計(jì)算的正確性。

from operator import add
sc.parallelize([1, 2, 3, 4, 5]).reduce(add)
15

reduceByKey(func, numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)

from operator import add
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
sorted(rdd.reduceByKey(add).collect())
[('a', 3), ('b', 1)]

reduceByKeyLocally(func)

from operator import add
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
rdd.reduceByKeyLocally(add)
{'a': 3, 'b': 1}

repartition(numPartitions)
repartitionAndSortWithinPartitions(numPartitions=None, partitionFunc=<function portable_hash at 0x7f51f1ac0668>, ascending=True, keyfunc=<function <lambda> at 0x7f51f1ab3ed8>)
根據(jù)給定的分區(qū)器對RDD進(jìn)行重新分區(qū)，在每個(gè)結(jié)果分區(qū)中，按照key值對記錄排序。這在每個(gè)分區(qū)中比先調(diào)用repartition再排序效率更高，因?yàn)樗梢詫⑴判蜻^程在shuffle操作的機(jī)器上進(jìn)行。
rightOuterJoin(other, numPartitions=None)
sample(withReplacement, fraction, seed=None)
對數(shù)據(jù)采樣。用戶可以設(shè)定是否有放回（withReplacement）、采樣的百分比（fraction）、隨機(jī)種子（seed）。
sampleByKey(withReplacement, fractions, seed=None)
sampleStdev()
計(jì)算RDD元素的樣本方差（通過除以N-1而不是N來估計(jì)方差中的偏差）。
saveAsHadoopDataset(conf, keyConverter=None, valueConverter=None)
saveAsHadoopFile(path, outputFormatClass, keyClass=None, valueClass=None, keyConverter=None, valueConverter=None, conf=None, compressionCodecClass=None)
saveAsNewAPIHadoopDataset(conf, keyConverter=None, valueConverter=None)
saveAsNewAPIHadoopFile(path, outputFormatClass, keyClass=None, valueClass=None, keyConverter=None, valueConverter=None, conf=None)
saveAsPickleFile(path, batchSize=10)
saveAsSequenceFile(path, compressionCodecClass=None)
saveAsTextFile(path, compressionCodecClass=None)
將此RDD保存為文本文件，使用元素的字符串表示形式。

sc.parallelize(range(10)).saveAsTextFile('file:/home/test')

setName(name)
為此RDD指定一個(gè)名稱。
sortBy(keyfunc, ascending=True, numPartitions=None)
按給定的keyfunc對此RDD進(jìn)行排序。
sortByKey(ascending=True, numPartitions=None, keyfunc=<function <lambda> at 0x7f51f1ab5050>)
stats()
返回一個(gè)StatCounter對象，在一次操作中捕獲RDD元素的均值，方差和計(jì)數(shù)。

sc.parallelize([1,2,3,4,5]).stats()
(count: 5, mean: 3.0, stdev: 1.41421356237, max: 5, min: 1)

stdev()
計(jì)算這個(gè)RDD元素的標(biāo)準(zhǔn)偏差。
subtract(other, numPartitions=None)
返回一個(gè)值，這個(gè)值包含在self中，且不包含在other中。

x = sc.parallelize([("a", 1), ("b", 4), ("b", 5), ("a", 3)])
y = sc.parallelize([("a", 3), ("c", None)])
sorted(x.subtract(y).collect())
[('a', 1), ('b', 4), ('b', 5)]

subtractByKey(other, numPartitions=None)
sum()
sumApprox(timeout, confidence=0.95)

實(shí)驗(yàn)階段

take(num)
返回RDD中前num個(gè)元素。
takeOrdered(num, key=None)
返回排序之后的前num個(gè)元素。
返回RDD按自然順序或自定義順序排序后的前n個(gè)元素。
takeSample(withReplacement, num, seed=None)
隨機(jī)取num個(gè)元素。withReplacement若為True表示有放回的取樣。
對一個(gè)數(shù)據(jù)集隨機(jī)抽樣，返回一個(gè)包含num個(gè)隨機(jī)抽樣元素的數(shù)組，參數(shù)withReplacement指定是否有放回抽樣，參數(shù)seed指定生成隨機(jī)數(shù)的種子。
toDebugString()
toLocalIterator()
返回包含此RDD中所有元素的迭代器。迭代器將消耗與RDD中最大分區(qū)一樣多的內(nèi)存。

rdd = sc.parallelize(range(10))
[x for x in rdd.toLocalIterator()]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

top(num, key=None)
treeAggregate(zeroValue, seqOp, combOp, depth=2)
以多級(jí)樹形模式聚合此RDD的元素。
treeReduce(f, depth=2)
union(other)

rdd = sc.parallelize([1, 1, 2, 3])
rdd.union(rdd).collect()
[1, 1, 2, 3, 1, 1, 2, 3]

unpersist()
將RDD標(biāo)記為非持久性，并從內(nèi)存和磁盤中刪除所有塊。
values()

m = sc.parallelize([(1, 2), (3, 4)]).values()
m.collect()
[2, 4]

variance()
計(jì)算RDD元素的方差。
zip(other)

x = sc.parallelize(range(0,5))
y = sc.parallelize(range(1000, 1005))
x.zip(y).collect()
[(0, 1000), (1, 1001), (2, 1002), (3, 1003), (4, 1004)]

zipWithIndex()
zipWithUniqueId()

class pyspark.StorageLevel(useDisk, useMemory, useOffHeap, deserialized, replication=1)

用于控制RDD存儲(chǔ)的標(biāo)志。每個(gè)StorageLevel記錄是否使用內(nèi)存，如果內(nèi)存不足，是否將RDD丟棄到磁盤，是否以特定于JAVA的序列化格式將數(shù)據(jù)保存在內(nèi)存中，以及是否在多個(gè)節(jié)點(diǎn)上復(fù)制RDD分區(qū)。還包含一些常用存儲(chǔ)級(jí)別MEMORY_ONLY的靜態(tài)常量。由于數(shù)據(jù)總是在Python端序列化，所有的常量使用序列化的格式。

DISK_ONLY = StorageLevel(True, False, False, False, 1)
DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)
MEMORY_AND_DISK = StorageLevel(True, True, False, False, 1)
MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)
MEMORY_AND_DISK_SER = StorageLevel(True, True, False, False, 1)
MEMORY_AND_DISK_SER_2 = StorageLevel(True, True, False, False, 2)
MEMORY_ONLY = StorageLevel(False, True, False, False, 1)
MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)
MEMORY_ONLY_SER = StorageLevel(False, True, False, False, 1)
MEMORY_ONLY_SER_2 = StorageLevel(False, True, False, False, 2)
OFF_HEAP = StorageLevel(True, True, True, False, 1)

class pyspark.Broadcast(sc=None, value=None, pickle_registry=None, path=None)

使用SparkContext.broadcast（）創(chuàng)建的廣播變量。通過value方法獲取值。

destroy()
銷毀與此廣播變量相關(guān)的所有數(shù)據(jù)和元數(shù)據(jù)。謹(jǐn)慎使用; 廣播變量一旦被銷毀，就不能再使用。這個(gè)方法阻塞直到銷毀完成。
dump(value, f)
load(path)
unpersist(blocking=False)
在每個(gè)executor上刪除這個(gè)廣播的緩存副本。如果在調(diào)用之后使用廣播，則需要將其重新發(fā)送給每個(gè)executor。
value
返回廣播的值

class pyspark.Accumulator(aid, value, accum_param)

可以累積的共享變量，即具有可交換和關(guān)聯(lián)的“add”操作。 Spark群集上的worker tasks可以使用+ =操作符將值添加到累加器，但只有驅(qū)動(dòng)程序可以訪問其值。 workers的更新會(huì)自動(dòng)傳播給驅(qū)動(dòng)程序。
雖然SparkContext支持像int和float這樣的原始數(shù)據(jù)類型的累加器，但是用戶也可以通過提供一個(gè)自定義的AccumulatorParam對象來為自定義類型定義累加器。

add(term)
value
獲取累加器的值; 只能在driver program中使用。

class pyspark.AccumulatorParam

幫助對象，定義如何累積給定類型的值。

addInPlace(value1, value2)
添加累加器數(shù)據(jù)類型的兩個(gè)值，返回一個(gè)新值; 為了效率，也可以更新value1并返回它。
zero(value)

class pyspark.MarshalSerializer

使用Python的Marshal序列化器序列化對象:
http://docs.python.org/2/library/marshal.html
這個(gè)序列化器比PickleSerializer更快，但支持更少的數(shù)據(jù)類型。

dumps(obj)
loads(obj)

class pyspark.PickleSerializer

使用Python的pickle序列化器序列化對象：
http://docs.python.org/2/library/pickle.html
這個(gè)序列化程序支持幾乎所有的Python對象，但是可能不像更專用的序列化程序那么快。

dumps(obj)
loads(obj, encoding=None)

class pyspark.StatusTracker(jtracker)

用于monitoring job and stage progress的Low-level報(bào)告API。
這些API有意提供非常弱的一致性語義; 這些API的消費(fèi)者應(yīng)該準(zhǔn)備好處理空的/丟失的信息。例如，作業(yè)的階段id可能是已知的，但狀態(tài)API可能沒有關(guān)于這些階段的細(xì)節(jié)的任何信息，所以getStageInfo可能會(huì)返回?zé)o效的有效階段id。
為了限制內(nèi)存使用量，這些API只提供最近作業(yè)/階段的信息。這些API將提供最后一個(gè)spark.ui.retainedStages階段和spark.ui.retainedJobs作業(yè)的信息。

getActiveJobsIds()
返回包含所有活動(dòng)jobs的ids的數(shù)組。
getActiveStageIds()
返回一個(gè)包含所有活動(dòng)stages的id的數(shù)組。
getJobIdsForGroup(jobGroup=None)
返回特定工作組中所有已知作業(yè)的列表。如果jobGroup為None，則返回所有與作業(yè)組不相關(guān)的作業(yè)。
返回的列表可能包含正在運(yùn)行的，失敗的和已完成的作業(yè)，并且可能因該方法的調(diào)用而有所不同。該方法不保證結(jié)果中元素的順序。
getJobInfo(jobId)
返回一個(gè)SparkJobInfo對象，如果作業(yè)信息找不到或被垃圾回收，則返回None。
getStageInfo(stageId)
返回SparkStageInfo對象，如果找不到階段信息或垃圾收集，則返回None。

class pyspark.SparkJobInfo

公開有關(guān)Spark作業(yè)的信息。

class pyspark.SparkStageInfo

公開有關(guān)Spark Stage的信息。

class pyspark.Profiler(ctx)

PySpark支持自定義分析器，這是為了允許使用不同的分析器，以及輸出不同于BasicProfiler中提供的格式。

dump(id, path)
將配置文件轉(zhuǎn)儲(chǔ)到路徑中，id是RDD id。
profile(func)
做函數(shù)func的分析
show(id)
將配置文件統(tǒng)計(jì)信息打印到標(biāo)準(zhǔn)輸出，id是RDD id。
stats()
返回收集的性能分析統(tǒng)計(jì)信息（pstats.Stats）

class pyspark.BasicProfiler(ctx)

BasicProfiler是默認(rèn)的profiler，它是基于cProfile和Accumulator實(shí)現(xiàn)的。

profile(func)
運(yùn)行并分析傳入的方法to_profile。返回配置文件對象。
stats()

class pyspark.TaskContext

實(shí)驗(yàn)階段

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

Spark Python API Docs(part one)

Spark Python API Docs(part one)

pyspark package

subpackages

contents

class pyspark.SpackConf(loadDefaults=True,_jvm=None,_jconf=None)

class 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'>)

class pyspark.SparkFiles

class pyspark.RDD(jrdd, ctx, jrdd_deserializer=AutoBatchedSerializer(PickleSerializer()))

class pyspark.StorageLevel(useDisk, useMemory, useOffHeap, deserialized, replication=1)

class pyspark.Broadcast(sc=None, value=None, pickle_registry=None, path=None)

class pyspark.Accumulator(aid, value, accum_param)

class pyspark.AccumulatorParam

class pyspark.MarshalSerializer

class pyspark.PickleSerializer

class pyspark.StatusTracker(jtracker)

class pyspark.SparkJobInfo

class pyspark.SparkStageInfo

class pyspark.Profiler(ctx)

class pyspark.BasicProfiler(ctx)

class pyspark.TaskContext

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

Spark Python API Docs(part one)

pyspark package

subpackages

contents

class pyspark.SpackConf(loadDefaults=True,_jvm=None,_jconf=None)

class 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'>)

class pyspark.SparkFiles

class pyspark.RDD(jrdd, ctx, jrdd_deserializer=AutoBatchedSerializer(PickleSerializer()))

class pyspark.StorageLevel(useDisk, useMemory, useOffHeap, deserialized, replication=1)

class pyspark.Broadcast(sc=None, value=None, pickle_registry=None, path=None)

class pyspark.Accumulator(aid, value, accum_param)

class pyspark.AccumulatorParam

class pyspark.MarshalSerializer

class pyspark.PickleSerializer

class pyspark.StatusTracker(jtracker)

class pyspark.SparkJobInfo

class pyspark.SparkStageInfo

class pyspark.Profiler(ctx)

class pyspark.BasicProfiler(ctx)

class pyspark.TaskContext

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av