spark中三种数据存储方式
[spark三种数据存储格式之间的转化参考link]
RDD
Spark的核⼼心概念是RDD (resilient distributed dataset)，指的是一个只读的、 可分区的、分布式数据集，这个数据集的全部或部分可以缓存在内存中，在多次计算间重⽤用
DataSet
DataFrame
注意⚠️: DataFrame是Dataset[Row] 的别名
附录
在site-packages/pyspark/sql/dataframe.py里面可以看到类DataFrame的定义
class DataFrame(object): ...
def __init__(self, jdf, sql_ctx): ...
@property
@since(1.3)
def rdd(self): ...
@property
@since("1.3.1")
def na(self): ...
@property
@since(1.4)
def stat(self): ...
@ignore_unicode_prefix
@since(1.3)
def toJSON(self, use_unicode=True): ...
@since(1.3)
def registerTempTable(self, name): ...
@since(2.0)
def createTempView(self, name): ...
@since(2.0)
def createOrReplaceTempView(self, name): ...
@since(2.1)
def createGlobalTempView(self, name): ...
@since(2.2)
def createOrReplaceGlobalTempView(self, name): ...
@property
@since(1.4)
def write(self): ...
@property
@since(2.0)
def writeStream(self): ...
@property
@since(1.3)
def schema(self): ...
@since(1.3)
def printSchema(self): ...
@since(1.3)
def explain(self, extended=False): ...
@since(2.4)
def exceptAll(self, other): ...
@since(1.3)
def isLocal(self): ...
@property
@since(2.0)
def isStreaming(self): ...
@since(1.3)
def show(self, n=20, truncate=True, vertical=False): ...
def __repr__(self): ...
def _repr_html_(self): ...
@since(2.1)
def checkpoint(self, eager=True): ...
@since(2.3)
def localCheckpoint(self, eager=True): ...
@since(2.1)
def withWatermark(self, eventTime, delayThreshold): ...
@since(2.2)
def hint(self, name, *parameters): ...
@since(1.3)
def count(self): ...
@ignore_unicode_prefix
@since(1.3)
def collect(self): ...
@ignore_unicode_prefix
@since(2.0)
def toLocalIterator(self): ...
@ignore_unicode_prefix
@since(1.3)
def limit(self, num): ...
@ignore_unicode_prefix
@since(1.3)
def take(self, num): ...
@since(1.3)
def foreach(self, f): ...
@since(1.3)
def foreachPartition(self, f): ...
@since(1.3)
def cache(self): ...
@since(1.3)
def persist(self, storageLevel=StorageLevel.MEMORY_AND_DISK): ...
@property
@since(2.1)
def storageLevel(self): ...
@since(1.3)
def unpersist(self, blocking=False): ...
@since(1.4)
def coalesce(self, numPartitions): ...
@since(1.3)
def repartition(self, numPartitions, *cols): ...
@since("2.4.0")
def repartitionByRange(self, numPartitions, *cols): ...
@since(1.3)
def distinct(self): ...
@since(1.3)
def sample(self, withReplacement=None, fraction=None, seed=None): ...
@since(1.5)
def sampleBy(self, col, fractions, seed=None): ...
@since(1.4)
def randomSplit(self, weights, seed=None): ...
@property
@since(1.3)
def dtypes(self): ...
@property
@since(1.3)
def columns(self): ...
@since(2.3)
def colRegex(self, colName): ...
@ignore_unicode_prefix
@since(1.3)
def alias(self, alias): ...
@ignore_unicode_prefix
@since(2.1)
def crossJoin(self, other): ...
@ignore_unicode_prefix
@since(1.3)
def join(self, other, on=None, how=None): ...
@since(1.6)
def sortWithinPartitions(self, *cols, **kwargs): ...
@ignore_unicode_prefix
@since(1.3)
def sort(self, *cols, **kwargs): ...
orderBy = sort
def _jseq(self, cols, converter=None): ...
def _jmap(self, jm): ...
def _jcols(self, *cols): ...
def _sort_cols(self, cols, kwargs): ...
@since("1.3.1")
def describe(self, *cols): ...
@since("2.3.0")
def summary(self, *statistics): ...
@ignore_unicode_prefix
@since(1.3)
def head(self, n=None): ...
@ignore_unicode_prefix
@since(1.3)
def first(self): ...
@ignore_unicode_prefix
@since(1.3)
def __getitem__(self, item): ...
@since(1.3)
def __getattr__(self, name): ...
@ignore_unicode_prefix
@since(1.3)
def select(self, *cols): ...
@since(1.3)
def selectExpr(self, *expr): ...
@ignore_unicode_prefix
@since(1.3)
def filter(self, condition): ...
@ignore_unicode_prefix
@since(1.3)
def groupBy(self, *cols): ...
@since(1.4)
def rollup(self, *cols): ...
@since(1.4)
def cube(self, *cols): ...
@since(1.3)
def agg(self, *exprs): ...
@since(2.0)
def union(self, other): ...
@since(1.3)
def unionAll(self, other): ...
@since(2.3)
def unionByName(self, other): ...
@since(1.3)
def intersect(self, other): ...
@since(2.4)
def intersectAll(self, other): ...
@since(1.3)
def subtract(self, other): ...
@since(1.4)
def dropDuplicates(self, subset=None): ...
@since("1.3.1")
def dropna(self, how='any', thresh=None, subset=None): ...
@since("1.3.1")
def fillna(self, value, subset=None): ...
@since(1.4)
def replace(self, to_replace, value=_NoValue, subset=None): ...
@since(2.0)
def approxQuantile(self, col, probabilities, relativeError): ...
@since(1.4)
def corr(self, col1, col2, method=None): ...
@since(1.4)
def cov(self, col1, col2): ...
@since(1.4)
def crosstab(self, col1, col2): ...
@since(1.4)
def freqItems(self, cols, support=None): ...
@ignore_unicode_prefix
@since(1.3)
def withColumn(self, colName, col): ...
@ignore_unicode_prefix
@since(1.3)
def withColumnRenamed(self, existing, new): ...
@since(1.4)
@ignore_unicode_prefix
def drop(self, *cols): ...
@ignore_unicode_prefix
def toDF(self, *cols): ...
@since(1.3)
def toPandas(self): ...
def _collectAsArrow(self): ...
def _to_scala_map(sc, jm): ...
def _to_corrected_pandas_type(dt): ...
class DataFrameNaFunctions(object): ...
def __init__(self, df): ...
def drop(self, how='any', thresh=None, subset=None): ...
drop.__doc__ = DataFrame.dropna.__doc__
def fill(self, value, subset=None): ...
fill.__doc__ = DataFrame.fillna.__doc__
def replace(self, to_replace, value=_NoValue, subset=None): ...
replace.__doc__ = DataFrame.replace.__doc__
class DataFrameStatFunctions(object):...
def __init__(self, df): ...
def approxQuantile(self, col, probabilities, relativeError): ...
approxQuantile.__doc__ = DataFrame.approxQuantile.__doc__
def corr(self, col1, col2, method=None): ...
corr.__doc__ = DataFrame.corr.__doc__
def cov(self, col1, col2): ...
cov.__doc__ = DataFrame.cov.__doc__
def crosstab(self, col1, col2): ...
crosstab.__doc__ = DataFrame.crosstab.__doc__
def freqItems(self, cols, support=None): ...
freqItems.__doc__ = DataFrame.freqItems.__doc__
def sampleBy(self, col, fractions, seed=None): ...
sampleBy.__doc__ = DataFrame.sampleBy.__doc__
在site-packages/pyspark/storagelevel.py中可以看到类StorageLevel的定义
class StorageLevel(object): ...
def __init__(self, useDisk, useMemory, useOffHeap, deserialized, replication=1):
self.useDisk = useDisk
self.useMemory = useMemory
self.useOffHeap = useOffHeap
self.deserialized = deserialized
self.replication = replication
def __repr__(self): ...
def __str__(self): ...
StorageLevel.DISK_ONLY = StorageLevel(True, False, False, False)
StorageLevel.DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)
StorageLevel.MEMORY_ONLY = StorageLevel(False, True, False, False)
StorageLevel.MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)
StorageLevel.MEMORY_AND_DISK = StorageLevel(True, True, False, False)
StorageLevel.MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)
StorageLevel.OFF_HEAP = StorageLevel(True, True, True, False, 1)
在site-packages/pyspark/rdd.py里可以看到RDD类的定义
class RDD(object): ...
def __init__(self, jrdd, ctx, jrdd_deserializer=AutoBatchedSerializer(PickleSerializer())): ...
def _pickled(self): ...
def id(self): ...
def __repr__(self): ...
def __getnewargs__(self): ...
@property
def context(self): ...
def cache(self): ...
def persist(self, storageLevel=StorageLevel.MEMORY_ONLY): ...
def unpersist(self): ...
def checkpoint(self): ...
def isCheckpointed(self): ...
def localCheckpoint(self): ...
def isLocallyCheckpointed(self): ...
def getCheckpointFile(self): ...
def map(self, f, preservesPartitioning=False): ...
def flatMap(self, f, preservesPartitioning=False): ...
def mapPartitions(self, f, preservesPartitioning=False): ...
def mapPartitionsWithIndex(self, f, preservesPartitioning=False): ...
def mapPartitionsWithSplit(self, f, preservesPartitioning=False): ...
def getNumPartitions(self): ...
def filter(self, f): ...
def distinct(self, numPartitions=None): ...
def sample(self, withReplacement, fraction, seed=None): ...
def randomSplit(self, weights, seed=None): ...
# this is ported from scala/spark/RDD.scala
def takeSample(self, withReplacement, num, seed=None): ...
@staticmethod
def _computeFractionForSampleSize(sampleSizeLowerBound, total, withReplacement): ...
def union(self, other): ...
def intersection(self, other): ...
def _reserialize(self, serializer=None): ...
def __add__(self, other): ...
def repartitionAndSortWithinPartitions(self, numPartitions=None, partitionFunc=portable_hash,
ascending=True, keyfunc=lambda x: x): ...
def sortByKey(self, ascending=True, numPartitions=None, keyfunc=lambda x: x): ...
def sortBy(self, keyfunc, ascending=True, numPartitions=None): ...
def glom(self): ...
def cartesian(self, other): ...
def groupBy(self, f, numPartitions=None, partitionFunc=portable_hash): ...
@ignore_unicode_prefix
def pipe(self, command, env=None, checkCode=False): ...
def foreach(self, f): ...
def foreachPartition(self, f): ...
def collect(self): ...
def reduce(self, f): ...
def treeReduce(self, f, depth=2): ...
def fold(self, zeroValue, op): ...
def aggregate(self, zeroValue, seqOp, combOp): ...
def treeAggregate(self, zeroValue, seqOp, combOp, depth=2): ...
def max(self, key=None): ...
def min(self, key=None): ...
def sum(self): ...
def count(self): ...
def stats(self): ...
def histogram(self, buckets): ...
def mean(self): ...
def variance(self): ...
def stdev(self): ...
def sampleStdev(self): ...
def sampleVariance(self): ...
def countByValue(self): ...
def top(self, num, key=None): ...
def takeOrdered(self, num, key=None): ...
def take(self, num): ...
def first(self): ...
def isEmpty(self): ...
def saveAsNewAPIHadoopDataset(self, conf, keyConverter=None, valueConverter=None): ...
def saveAsNewAPIHadoopFile(self, path, outputFormatClass, keyClass=None, valueClass=None,
keyConverter=None, valueConverter=None, conf=None): ...
def saveAsHadoopDataset(self, conf, keyConverter=None, valueConverter=None): ...
def saveAsHadoopFile(self, path, outputFormatClass, keyClass=None, valueClass=None,
keyConverter=None, valueConverter=None, conf=None,
compressionCodecClass=None): ...
def saveAsSequenceFile(self, path, compressionCodecClass=None): ...
def saveAsPickleFile(self, path, batchSize=10): ...
@ignore_unicode_prefix
def saveAsTextFile(self, path, compressionCodecClass=None): ...
# Pair functions
def collectAsMap(self): ...
def keys(self): ...
def values(self): ...
def reduceByKey(self, func, numPartitions=None, partitionFunc=portable_hash): ...
def reduceByKeyLocally(self, func): ...
def countByKey(self): ...
def join(self, other, numPartitions=None): ...
def leftOuterJoin(self, other, numPartitions=None): ...
def rightOuterJoin(self, other, numPartitions=None): ...
def fullOuterJoin(self, other, numPartitions=None): ...
# TODO: add option to control map-side combining
# portable_hash is used as default, because builtin hash of None is different
# cross machines.
def partitionBy(self, numPartitions, partitionFunc=portable_hash): ...
# TODO: add control over map-side aggregation
def combineByKey(self, createCombiner, mergeValue, mergeCombiners,
numPartitions=None, partitionFunc=portable_hash): ...
def aggregateByKey(self, zeroValue, seqFunc, combFunc, numPartitions=None,
partitionFunc=portable_hash): ...
def foldByKey(self, zeroValue, func, numPartitions=None, partitionFunc=portable_hash): ...
def _memory_limit(self): ...
# TODO: support variant with custom partitioner
def groupByKey(self, numPartitions=None, partitionFunc=portable_hash): ...
def flatMapValues(self, f): ...
def mapValues(self, f): ...
def groupWith(self, other, *others): ...
# TODO: add variant with custom parittioner
def cogroup(self, other, numPartitions=None): ...
def sampleByKey(self, withReplacement, fractions, seed=None): ...
def subtractByKey(self, other, numPartitions=None): ...
def subtract(self, other, numPartitions=None): ...
def keyBy(self, f): ...
def repartition(self, numPartitions): ...
def coalesce(self, numPartitions, shuffle=False): ...
def zip(self, other): ...
def zipWithIndex(self): ...
def zipWithUniqueId(self): ...
def name(self): ...
@ignore_unicode_prefix
def setName(self, name): ...
def toDebugString(self): ...
def getStorageLevel(self): ...
def _defaultReducePartitions(self): ...
def lookup(self, key): ...
def _to_java_object_rdd(self): ...
def countApprox(self, timeout, confidence=0.95): ...
def sumApprox(self, timeout, confidence=0.95): ...
def meanApprox(self, timeout, confidence=0.95): ...
def countApproxDistinct(self, relativeSD=0.05): ...
def toLocalIterator(self): ...
def barrier(self): ...
def _is_barrier(self): ...
