Coverage for pyspark/rdd.py : 94%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
|
# # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. #
CartesianDeserializer, CloudPickleSerializer, PairDeserializer, PickleSerializer, \ pack_long, read_int, write_int python_right_outer_join, python_full_outer_join, python_cogroup get_used_memory, ExternalSorter, ExternalGroupBy
""" Evaluation type of python rdd.
These values are internal to PySpark.
These values should match values in org.apache.spark.api.python.PythonEvalType. """
""" This function returns consistent hash code for builtin types, especially for None and tuple with None.
The algorithm is similar to that one used by CPython 2.7
Examples -------- >>> portable_hash(None) 0 >>> portable_hash((None, 1)) & 0xffffffff 219750521 """
raise RuntimeError("Randomness of hash of string should be disabled via PYTHONHASHSEED")
h = -2
""" Bounded value is generated by approximate job, with confidence and low bound and high bound.
Examples -------- >>> BoundedFloat(100.0, 0.95, 95.0, 105.0) 100.0 """
""" Create a local socket that can be used to load deserialized data from the JVM
Parameters ---------- sock_info : tuple Tuple containing port number and authentication secret for a local socket.
Returns ------- sockfile file descriptor of the local socket """ # The RDD materialization time is unpredictable, if we set a timeout for socket reading # operation, it will very possibly fail. See SPARK-18281.
""" Connect to a local socket described by sock_info and use the given serializer to yield data
Parameters ---------- sock_info : tuple Tuple containing port number and authentication secret for a local socket. serializer : :py:class:`Serializer` The PySpark serializer to use
Returns ------- result of :py:meth:`Serializer.load_stream`, usually a generator that yields deserialized data """ # The socket will be automatically closed when garbage-collected.
""" Create a synchronous local iterable over a socket """
# Request next partition data from Java
# If response is 1 then there is a partition to read, if 0 then fully consumed
# Load the partition data as a stream and read each item
# An error occurred, join serving thread and raise any exceptions from the JVM
# If local iterator is not fully consumed, # Finish consuming partition data stream # Tell Java to stop sending data and close connection except Exception: # Ignore any errors, socket is automatically closed when garbage-collected pass
and self.partitionFunc == other.partitionFunc)
""" A Resilient Distributed Dataset (RDD), the basic abstraction in Spark. Represents an immutable, partitioned collection of elements that can be operated on in parallel. """
""" A unique ID for this RDD (within its SparkContext). """ return self._id
return self._jrdd.toString()
# This method is called when attempting to pickle an RDD, which is always an error: raise RuntimeError( "It appears that you are attempting to broadcast an RDD or reference an RDD from an " "action or transformation. RDD transformations and actions can only be invoked by the " "driver, not inside of other transformations; for example, " "rdd1.map(lambda x: rdd2.values.count() * x) is invalid because the values " "transformation and count action cannot be performed inside of the rdd1.map " "transformation. For more information, see SPARK-5063." )
def context(self): """ The :class:`SparkContext` that this RDD was created on. """
""" Persist this RDD with the default storage level (`MEMORY_ONLY`). """
""" Set this RDD's storage level to persist its values across operations after the first time it is computed. This can only be used to assign a new storage level if the RDD does not have a storage level set yet. If no storage level is specified defaults to (`MEMORY_ONLY`).
Examples -------- >>> rdd = sc.parallelize(["b", "a", "c"]) >>> rdd.persist().is_cached True """
""" Mark the RDD as non-persistent, and remove all blocks for it from memory and disk.
.. versionchanged:: 3.0.0 Added optional argument `blocking` to specify whether to block until all blocks are deleted. """ self.is_cached = False self._jrdd.unpersist(blocking) return self
""" Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint directory set with :meth:`SparkContext.setCheckpointDir` and all references to its parent RDDs will be removed. This function must be called before any job has been executed on this RDD. It is strongly recommended that this RDD is persisted in memory, otherwise saving it on a file will require recomputation. """
""" Return whether this RDD is checkpointed and materialized, either reliably or locally. """
""" Mark this RDD for local checkpointing using Spark's existing caching layer.
This method is for users who wish to truncate RDD lineages while skipping the expensive step of replicating the materialized data in a reliable distributed file system. This is useful for RDDs with long lineages that need to be truncated periodically (e.g. GraphX).
Local checkpointing sacrifices fault-tolerance for performance. In particular, checkpointed data is written to ephemeral local storage in the executors instead of to a reliable, fault-tolerant storage. The effect is that if an executor fails during the computation, the checkpointed data may no longer be accessible, causing an irrecoverable job failure.
This is NOT safe to use with dynamic allocation, which removes executors along with their cached blocks. If you must use both features, you are advised to set `spark.dynamicAllocation.cachedExecutorIdleTimeout` to a high value.
The checkpoint directory set through :meth:`SparkContext.setCheckpointDir` is not used. """
""" Return whether this RDD is marked for local checkpointing.
Exposed for testing. """
""" Gets the name of the file to which this RDD was checkpointed
Not defined if RDD is checkpointed locally. """
""" Return a new RDD by applying a function to each element of this RDD.
Examples -------- >>> rdd = sc.parallelize(["b", "a", "c"]) >>> sorted(rdd.map(lambda x: (x, 1)).collect()) [('a', 1), ('b', 1), ('c', 1)] """
""" Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.
Examples -------- >>> rdd = sc.parallelize([2, 3, 4]) >>> sorted(rdd.flatMap(lambda x: range(1, x)).collect()) [1, 1, 1, 2, 2, 3] >>> sorted(rdd.flatMap(lambda x: [(x, x), (x, x)]).collect()) [(2, 2), (2, 2), (3, 3), (3, 3), (4, 4), (4, 4)] """
""" Return a new RDD by applying a function to each partition of this RDD.
Examples -------- >>> rdd = sc.parallelize([1, 2, 3, 4], 2) >>> def f(iterator): yield sum(iterator) >>> rdd.mapPartitions(f).collect() [3, 7] """
""" Return a new RDD by applying a function to each partition of this RDD, while tracking the index of the original partition.
Examples -------- >>> rdd = sc.parallelize([1, 2, 3, 4], 4) >>> def f(splitIndex, iterator): yield splitIndex >>> rdd.mapPartitionsWithIndex(f).sum() 6 """
"""
Return a new RDD by applying a function to each partition of this RDD, while tracking the index of the original partition.
.. deprecated:: 0.9.0 use :py:meth:`RDD.mapPartitionsWithIndex` instead.
Examples -------- >>> rdd = sc.parallelize([1, 2, 3, 4], 4) >>> def f(splitIndex, iterator): yield splitIndex >>> rdd.mapPartitionsWithSplit(f).sum() 6 """ "mapPartitionsWithSplit is deprecated; use mapPartitionsWithIndex instead", FutureWarning, stacklevel=2 )
""" Returns the number of partitions in RDD
Examples -------- >>> rdd = sc.parallelize([1, 2, 3, 4], 2) >>> rdd.getNumPartitions() 2 """
""" Return a new RDD containing only the elements that satisfy a predicate.
Examples -------- >>> rdd = sc.parallelize([1, 2, 3, 4, 5]) >>> rdd.filter(lambda x: x % 2 == 0).collect() [2, 4] """
""" Return a new RDD containing the distinct elements in this RDD.
Examples -------- >>> sorted(sc.parallelize([1, 1, 2, 3]).distinct().collect()) [1, 2, 3] """ .reduceByKey(lambda x, _: x, numPartitions) \ .map(lambda x: x[0])
""" Return a sampled subset of this RDD.
Parameters ---------- withReplacement : bool can elements be sampled multiple times (replaced when sampled out) fraction : float expected size of the sample as a fraction of this RDD's size without replacement: probability that each element is chosen; fraction must be [0, 1] with replacement: expected number of times each element is chosen; fraction must be >= 0 seed : int, optional seed for the random number generator
Notes ----- This is not guaranteed to provide exactly the fraction specified of the total count of the given :class:`DataFrame`.
Examples -------- >>> rdd = sc.parallelize(range(100), 4) >>> 6 <= rdd.sample(False, 0.1, 81).count() <= 14 True """
""" Randomly splits this RDD with the provided weights.
weights : list weights for splits, will be normalized if they don't sum to 1 seed : int, optional random seed
Returns ------- list split RDDs in a list
Examples -------- >>> rdd = sc.parallelize(range(500), 1) >>> rdd1, rdd2 = rdd.randomSplit([2, 3], 17) >>> len(rdd1.collect() + rdd2.collect()) 500 >>> 150 < rdd1.count() < 250 True >>> 250 < rdd2.count() < 350 True """ seed = random.randint(0, 2 ** 32 - 1) for lb, ub in zip(cweights, cweights[1:])]
# this is ported from scala/spark/RDD.scala """ Return a fixed-size sampled subset of this RDD.
Notes ----- This method should only be used if the resulting array is expected to be small, as all the data is loaded into the driver's memory.
Examples -------- >>> rdd = sc.parallelize(range(0, 10)) >>> len(rdd.takeSample(True, 20, 1)) 20 >>> len(rdd.takeSample(False, 5, 2)) 5 >>> len(rdd.takeSample(False, 15, 3)) 10 """
raise ValueError("Sample size cannot be negative.") return []
return []
# shuffle current RDD and return
raise ValueError( "Sample size cannot be greater than %d." % maxSampleSize)
num, initialCount, withReplacement)
# If the first sample didn't turn out large enough, keep trying to take samples; # this shouldn't happen often because we use a big multiplier for their initial size. # See: scala/spark/RDD.scala # TODO: add log warning for when more than one iteration was run seed = rand.randint(0, sys.maxsize) samples = self.sample(withReplacement, fraction, seed).collect()
def _computeFractionForSampleSize(sampleSizeLowerBound, total, withReplacement): """ Returns a sampling rate that guarantees a sample of size >= sampleSizeLowerBound 99.99% of the time.
How the sampling rate is determined: Let p = num / total, where num is the sample size and total is the total number of data points in the RDD. We're trying to compute q > p such that - when sampling with replacement, we're drawing each data point with prob_i ~ Pois(q), where we want to guarantee Pr[s < num] < 0.0001 for s = sum(prob_i for i from 0 to total), i.e. the failure rate of not having a sufficiently large sample < 0.0001. Setting q = p + 5 * sqrt(p/total) is sufficient to guarantee 0.9999 success rate for num > 12, but we need a slightly larger q (9 empirically determined). - when sampling without replacement, we're drawing each data point with prob_i ~ Binomial(total, fraction) and our choice of q guarantees 1-delta, or 0.9999 success rate, where success rate is defined the same as in sampling with replacement. """ numStDev = 9 else:
""" Return the union of this RDD and another one.
Examples -------- >>> rdd = sc.parallelize([1, 1, 2, 3]) >>> rdd.union(rdd).collect() [1, 1, 2, 3, 1, 1, 2, 3] """ self._jrdd_deserializer) else: # These RDDs contain data in different serialized formats, so we # must normalize them to the default serializer. self.ctx.serializer) self.getNumPartitions() == rdd.getNumPartitions()):
""" Return the intersection of this RDD and another one. The output will not contain any duplicate elements, even if the input RDDs did.
Notes ----- This method performs a shuffle internally.
Examples -------- >>> 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] """ .cogroup(other.map(lambda v: (v, None))) \ .filter(lambda k_vs: all(k_vs[1])) \ .keys()
""" Return the union of this RDD and another one.
Examples -------- >>> rdd = sc.parallelize([1, 1, 2, 3]) >>> (rdd + rdd).collect() [1, 1, 2, 3, 1, 1, 2, 3] """ raise TypeError
ascending=True, keyfunc=lambda x: x): """ Repartition the RDD according to the given partitioner and, within each resulting partition, sort records by their keys.
Examples -------- >>> rdd = sc.parallelize([(0, 5), (3, 8), (2, 6), (0, 8), (3, 8), (1, 3)]) >>> rdd2 = rdd.repartitionAndSortWithinPartitions(2, lambda x: x % 2, True) >>> rdd2.glom().collect() [[(0, 5), (0, 8), (2, 6)], [(1, 3), (3, 8), (3, 8)]] """ numPartitions = self._defaultReducePartitions()
""" Sorts this RDD, which is assumed to consist of (key, value) pairs.
Examples -------- >>> tmp = [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)] >>> sc.parallelize(tmp).sortByKey().first() ('1', 3) >>> sc.parallelize(tmp).sortByKey(True, 1).collect() [('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)] >>> sc.parallelize(tmp).sortByKey(True, 2).collect() [('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)] >>> tmp2 = [('Mary', 1), ('had', 2), ('a', 3), ('little', 4), ('lamb', 5)] >>> tmp2.extend([('whose', 6), ('fleece', 7), ('was', 8), ('white', 9)]) >>> sc.parallelize(tmp2).sortByKey(True, 3, keyfunc=lambda k: k.lower()).collect() [('a', 3), ('fleece', 7), ('had', 2), ('lamb', 5),...('white', 9), ('whose', 6)] """
# first compute the boundary of each part via sampling: we want to partition # the key-space into bins such that the bins have roughly the same # number of (key, value) pairs falling into them
# we have numPartitions many parts but one of the them has # an implicit boundary for i in range(0, numPartitions - 1)]
else:
""" Sorts this RDD by the given keyfunc
Examples -------- >>> tmp = [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)] >>> sc.parallelize(tmp).sortBy(lambda x: x[0]).collect() [('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)] >>> sc.parallelize(tmp).sortBy(lambda x: x[1]).collect() [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)] """
""" Return an RDD created by coalescing all elements within each partition into a list.
Examples -------- >>> rdd = sc.parallelize([1, 2, 3, 4], 2) >>> sorted(rdd.glom().collect()) [[1, 2], [3, 4]] """
""" Return the Cartesian product of this RDD and another one, that is, the RDD of all pairs of elements ``(a, b)`` where ``a`` is in `self` and ``b`` is in `other`.
Examples -------- >>> rdd = sc.parallelize([1, 2]) >>> sorted(rdd.cartesian(rdd).collect()) [(1, 1), (1, 2), (2, 1), (2, 2)] """ # Due to batching, we can't use the Java cartesian method. other._jrdd_deserializer)
""" Return an RDD of grouped items.
Examples -------- >>> 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])] """
""" Return an RDD created by piping elements to a forked external process.
Parameters ---------- command : str command to run. env : dict, optional environment variables to set. checkCode : bool, optional whether or not to check the return value of the shell command.
Examples -------- >>> sc.parallelize(['1', '2', '', '3']).pipe('cat').collect() ['1', '2', '', '3'] """
shlex.split(command), env=env, stdin=PIPE, stdout=PIPE)
"with error code %d" % (command, pipe.returncode)) else: yield i chain(iter(pipe.stdout.readline, b''), check_return_code()))
""" Applies a function to all elements of this RDD.
Examples -------- >>> def f(x): print(x) >>> sc.parallelize([1, 2, 3, 4, 5]).foreach(f) """
""" Applies a function to each partition of this RDD.
Examples -------- >>> def f(iterator): ... for x in iterator: ... print(x) >>> sc.parallelize([1, 2, 3, 4, 5]).foreachPartition(f) """
""" Return a list that contains all of the elements in this RDD.
Notes ----- This method should only be used if the resulting array is expected to be small, as all the data is loaded into the driver's memory. """
""" When collect rdd, use this method to specify job group.
.. versionadded:: 3.0.0 .. deprecated:: 3.1.0 Use :class:`pyspark.InheritableThread` with the pinned thread mode enabled. """ "Deprecated in 3.1, Use pyspark.InheritableThread with " "the pinned thread mode enabled.", FutureWarning )
self._jrdd.rdd(), groupId, description, interruptOnCancel)
""" Reduces the elements of this RDD using the specified commutative and associative binary operator. Currently reduces partitions locally.
Examples -------- >>> from operator import add >>> sc.parallelize([1, 2, 3, 4, 5]).reduce(add) 15 >>> sc.parallelize((2 for _ in range(10))).map(lambda x: 1).cache().reduce(add) 10 >>> sc.parallelize([]).reduce(add) Traceback (most recent call last): ... ValueError: Can not reduce() empty RDD """
""" Reduces the elements of this RDD in a multi-level tree pattern.
Parameters ---------- f : function depth : int, optional suggested depth of the tree (default: 2)
Examples -------- >>> add = lambda x, y: x + y >>> rdd = sc.parallelize([-5, -4, -3, -2, -1, 1, 2, 3, 4], 10) >>> rdd.treeReduce(add) -5 >>> rdd.treeReduce(add, 1) -5 >>> rdd.treeReduce(add, 2) -5 >>> rdd.treeReduce(add, 5) -5 >>> rdd.treeReduce(add, 10) -5 """ raise ValueError("Depth cannot be smaller than 1 but got %d." % depth)
return x else:
raise ValueError("Cannot reduce empty RDD.")
""" Aggregate the elements of each partition, and then the results for all the partitions, using a given associative function and a neutral "zero value."
The function ``op(t1, t2)`` is allowed to modify ``t1`` and return it as its result value to avoid object allocation; however, it should not modify ``t2``.
This behaves somewhat differently from fold operations implemented for non-distributed collections in functional languages like Scala. This fold operation may be applied to partitions individually, and then fold those results into the final result, rather than apply the fold to each element sequentially in some defined ordering. For functions that are not commutative, the result may differ from that of a fold applied to a non-distributed collection.
Examples -------- >>> from operator import add >>> sc.parallelize([1, 2, 3, 4, 5]).fold(0, add) 15 """
# collecting result of mapPartitions here ensures that the copy of # zeroValue provided to each partition is unique from the one provided # to the final reduce call
""" Aggregate the elements of each partition, and then the results for all the partitions, using a given combine functions and a neutral "zero value."
The functions ``op(t1, t2)`` is allowed to modify ``t1`` and return it as its result value to avoid object allocation; however, it should not modify ``t2``.
The first function (seqOp) can return a different result type, U, than the type of this RDD. Thus, we need one operation for merging a T into an U and one operation for merging two U
Examples -------- >>> 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) >>> sc.parallelize([]).aggregate((0, 0), seqOp, combOp) (0, 0) """
# collecting result of mapPartitions here ensures that the copy of # zeroValue provided to each partition is unique from the one provided # to the final reduce call
""" Aggregates the elements of this RDD in a multi-level tree pattern.
depth : int, optional suggested depth of the tree (default: 2)
Examples -------- >>> add = lambda x, y: x + y >>> rdd = sc.parallelize([-5, -4, -3, -2, -1, 1, 2, 3, 4], 10) >>> rdd.treeAggregate(0, add, add) -5 >>> rdd.treeAggregate(0, add, add, 1) -5 >>> rdd.treeAggregate(0, add, add, 2) -5 >>> rdd.treeAggregate(0, add, add, 5) -5 >>> rdd.treeAggregate(0, add, add, 10) -5 """ raise ValueError("Depth cannot be smaller than 1 but got %d." % depth)
return zeroValue
# If creating an extra level doesn't help reduce the wall-clock time, we stop the tree # aggregation.
.mapPartitionsWithIndex(mapPartition) \ .reduceByKey(combOp, curNumPartitions) \ .values()
""" Find the maximum item in this RDD.
Parameters ---------- key : function, optional A function used to generate key for comparing
Examples -------- >>> rdd = sc.parallelize([1.0, 5.0, 43.0, 10.0]) >>> rdd.max() 43.0 >>> rdd.max(key=str) 5.0 """
""" Find the minimum item in this RDD.
Parameters ---------- key : function, optional A function used to generate key for comparing
Examples -------- >>> rdd = sc.parallelize([2.0, 5.0, 43.0, 10.0]) >>> rdd.min() 2.0 >>> rdd.min(key=str) 10.0 """
""" Add up the elements in this RDD.
Examples -------- >>> sc.parallelize([1.0, 2.0, 3.0]).sum() 6.0 """
""" Return the number of elements in this RDD.
Examples -------- >>> sc.parallelize([2, 3, 4]).count() 3 """
""" Return a :class:`StatCounter` object that captures the mean, variance and count of the RDD's elements in one operation. """
""" Compute a histogram using the provided buckets. The buckets are all open to the right except for the last which is closed. e.g. [1,10,20,50] means the buckets are [1,10) [10,20) [20,50], which means 1<=x<10, 10<=x<20, 20<=x<=50. And on the input of 1 and 50 we would have a histogram of 1,0,1.
If your histogram is evenly spaced (e.g. [0, 10, 20, 30]), this can be switched from an O(log n) insertion to O(1) per element (where n is the number of buckets).
Buckets must be sorted, not contain any duplicates, and have at least two elements.
If `buckets` is a number, it will generate buckets which are evenly spaced between the minimum and maximum of the RDD. For example, if the min value is 0 and the max is 100, given `buckets` as 2, the resulting buckets will be [0,50) [50,100]. `buckets` must be at least 1. An exception is raised if the RDD contains infinity. If the elements in the RDD do not vary (max == min), a single bucket will be used.
The return value is a tuple of buckets and histogram.
Examples -------- >>> rdd = sc.parallelize(range(51)) >>> rdd.histogram(2) ([0, 25, 50], [25, 26]) >>> rdd.histogram([0, 5, 25, 50]) ([0, 5, 25, 50], [5, 20, 26]) >>> rdd.histogram([0, 15, 30, 45, 60]) # evenly spaced buckets ([0, 15, 30, 45, 60], [15, 15, 15, 6]) >>> rdd = sc.parallelize(["ab", "ac", "b", "bd", "ef"]) >>> rdd.histogram(("a", "b", "c")) (('a', 'b', 'c'), [2, 2]) """
# filter out non-comparable elements return False
# faster than stats() if " empty " in str(e): raise ValueError("can not generate buckets from empty RDD") raise
# keep them as integer if possible
raise ValueError("can not have None or NaN in buckets")
raise ValueError("buckets should be sorted")
raise ValueError("buckets should not contain duplicated values")
else:
else:
else bisect.bisect_right(buckets, i) - 1) # add last two together
""" Compute the mean of this RDD's elements.
Examples -------- >>> sc.parallelize([1, 2, 3]).mean() 2.0 """
""" Compute the variance of this RDD's elements.
Examples -------- >>> sc.parallelize([1, 2, 3]).variance() 0.666... """
""" Compute the standard deviation of this RDD's elements.
Examples -------- >>> sc.parallelize([1, 2, 3]).stdev() 0.816... """
""" Compute the sample standard deviation of this RDD's elements (which corrects for bias in estimating the standard deviation by dividing by N-1 instead of N).
Examples -------- >>> sc.parallelize([1, 2, 3]).sampleStdev() 1.0 """
""" Compute the sample variance of this RDD's elements (which corrects for bias in estimating the variance by dividing by N-1 instead of N).
Examples -------- >>> sc.parallelize([1, 2, 3]).sampleVariance() 1.0 """
""" Return the count of each unique value in this RDD as a dictionary of (value, count) pairs.
Examples -------- >>> sorted(sc.parallelize([1, 2, 1, 2, 2], 2).countByValue().items()) [(1, 2), (2, 3)] """
""" Get the top N elements from an RDD.
Notes ----- This method should only be used if the resulting array is expected to be small, as all the data is loaded into the driver's memory.
It returns the list sorted in descending order.
Examples -------- >>> sc.parallelize([10, 4, 2, 12, 3]).top(1) [12] >>> sc.parallelize([2, 3, 4, 5, 6], 2).top(2) [6, 5] >>> sc.parallelize([10, 4, 2, 12, 3]).top(3, key=str) [4, 3, 2] """
""" Get the N elements from an RDD ordered in ascending order or as specified by the optional key function.
Notes ----- This method should only be used if the resulting array is expected to be small, as all the data is loaded into the driver's memory.
Examples -------- >>> sc.parallelize([10, 1, 2, 9, 3, 4, 5, 6, 7]).takeOrdered(6) [1, 2, 3, 4, 5, 6] >>> sc.parallelize([10, 1, 2, 9, 3, 4, 5, 6, 7], 2).takeOrdered(6, key=lambda x: -x) [10, 9, 7, 6, 5, 4] """
""" Take the first num elements of the RDD.
It works by first scanning one partition, and use the results from that partition to estimate the number of additional partitions needed to satisfy the limit.
Translated from the Scala implementation in RDD#take().
Notes ----- This method should only be used if the resulting array is expected to be small, as all the data is loaded into the driver's memory.
Examples -------- >>> sc.parallelize([2, 3, 4, 5, 6]).cache().take(2) [2, 3] >>> sc.parallelize([2, 3, 4, 5, 6]).take(10) [2, 3, 4, 5, 6] >>> sc.parallelize(range(100), 100).filter(lambda x: x > 90).take(3) [91, 92, 93] """
# The number of partitions to try in this iteration. # It is ok for this number to be greater than totalParts because # we actually cap it at totalParts in runJob. # If we didn't find any rows after the previous iteration, # quadruple and retry. Otherwise, interpolate the number of # partitions we need to try, but overestimate it by 50%. # We also cap the estimation in the end. else: # the first parameter of max is >=1 whenever partsScanned >= 2
""" Return the first element in this RDD.
Examples -------- >>> sc.parallelize([2, 3, 4]).first() 2 >>> sc.parallelize([]).first() Traceback (most recent call last): ... ValueError: RDD is empty """
""" Returns true if and only if the RDD contains no elements at all.
Notes ----- An RDD may be empty even when it has at least 1 partition.
Examples -------- >>> sc.parallelize([]).isEmpty() True >>> sc.parallelize([1]).isEmpty() False """
""" Output a Python RDD of key-value pairs (of form ``RDD[(K, V)]``) to any Hadoop file system, using the new Hadoop OutputFormat API (mapreduce package). Keys/values are converted for output using either user specified converters or, by default, "org.apache.spark.api.python.JavaToWritableConverter".
Parameters ---------- conf : dict Hadoop job configuration keyConverter : str, optional fully qualified classname of key converter (None by default) valueConverter : str, optional fully qualified classname of value converter (None by default) """ keyConverter, valueConverter, True)
keyConverter=None, valueConverter=None, conf=None): """ Output a Python RDD of key-value pairs (of form ``RDD[(K, V)]``) to any Hadoop file system, using the new Hadoop OutputFormat API (mapreduce package). Key and value types will be inferred if not specified. Keys and values are converted for output using either user specified converters or "org.apache.spark.api.python.JavaToWritableConverter". The `conf` is applied on top of the base Hadoop conf associated with the SparkContext of this RDD to create a merged Hadoop MapReduce job configuration for saving the data.
path : str path to Hadoop file outputFormatClass : str fully qualified classname of Hadoop OutputFormat (e.g. "org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat") keyClass : str, optional fully qualified classname of key Writable class (e.g. "org.apache.hadoop.io.IntWritable", None by default) valueClass : str, optional fully qualified classname of value Writable class (e.g. "org.apache.hadoop.io.Text", None by default) keyConverter : str, optional fully qualified classname of key converter (None by default) valueConverter : str, optional fully qualified classname of value converter (None by default) conf : dict, optional Hadoop job configuration (None by default) """ outputFormatClass, keyClass, valueClass, keyConverter, valueConverter, jconf)
""" Output a Python RDD of key-value pairs (of form ``RDD[(K, V)]``) to any Hadoop file system, using the old Hadoop OutputFormat API (mapred package). Keys/values are converted for output using either user specified converters or, by default, "org.apache.spark.api.python.JavaToWritableConverter".
Parameters ---------- conf : dict Hadoop job configuration keyConverter : str, optional fully qualified classname of key converter (None by default) valueConverter : str, optional fully qualified classname of value converter (None by default) """ keyConverter, valueConverter, False)
keyConverter=None, valueConverter=None, conf=None, compressionCodecClass=None): """ Output a Python RDD of key-value pairs (of form ``RDD[(K, V)]``) to any Hadoop file system, using the old Hadoop OutputFormat API (mapred package). Key and value types will be inferred if not specified. Keys and values are converted for output using either user specified converters or "org.apache.spark.api.python.JavaToWritableConverter". The `conf` is applied on top of the base Hadoop conf associated with the SparkContext of this RDD to create a merged Hadoop MapReduce job configuration for saving the data.
Parameters ---------- path : str path to Hadoop file outputFormatClass : str fully qualified classname of Hadoop OutputFormat (e.g. "org.apache.hadoop.mapred.SequenceFileOutputFormat") keyClass : str, optional fully qualified classname of key Writable class (e.g. "org.apache.hadoop.io.IntWritable", None by default) valueClass : str, optional fully qualified classname of value Writable class (e.g. "org.apache.hadoop.io.Text", None by default) keyConverter : str, optional fully qualified classname of key converter (None by default) valueConverter : str, optional fully qualified classname of value converter (None by default) conf : dict, optional (None by default) compressionCodecClass : str fully qualified classname of the compression codec class i.e. "org.apache.hadoop.io.compress.GzipCodec" (None by default) """ outputFormatClass, keyClass, valueClass, keyConverter, valueConverter, jconf, compressionCodecClass)
""" Output a Python RDD of key-value pairs (of form ``RDD[(K, V)]``) to any Hadoop file system, using the "org.apache.hadoop.io.Writable" types that we convert from the RDD's key and value types. The mechanism is as follows:
1. Pyrolite is used to convert pickled Python RDD into RDD of Java objects. 2. Keys and values of this Java RDD are converted to Writables and written out.
Parameters ---------- path : str path to sequence file compressionCodecClass : str, optional fully qualified classname of the compression codec class i.e. "org.apache.hadoop.io.compress.GzipCodec" (None by default) """ path, compressionCodecClass)
""" Save this RDD as a SequenceFile of serialized objects. The serializer used is :class:`pyspark.serializers.PickleSerializer`, default batch size is 10.
Examples -------- >>> from tempfile import NamedTemporaryFile >>> tmpFile = NamedTemporaryFile(delete=True) >>> tmpFile.close() >>> sc.parallelize([1, 2, 'spark', 'rdd']).saveAsPickleFile(tmpFile.name, 3) >>> sorted(sc.pickleFile(tmpFile.name, 5).map(str).collect()) ['1', '2', 'rdd', 'spark'] """ ser = AutoBatchedSerializer(PickleSerializer()) else:
""" Save this RDD as a text file, using string representations of elements.
Parameters ---------- path : str path to text file compressionCodecClass : str, optional fully qualified classname of the compression codec class i.e. "org.apache.hadoop.io.compress.GzipCodec" (None by default)
Examples -------- >>> from tempfile import NamedTemporaryFile >>> tempFile = NamedTemporaryFile(delete=True) >>> tempFile.close() >>> sc.parallelize(range(10)).saveAsTextFile(tempFile.name) >>> from fileinput import input >>> from glob import glob >>> ''.join(sorted(input(glob(tempFile.name + "/part-0000*")))) '0\\n1\\n2\\n3\\n4\\n5\\n6\\n7\\n8\\n9\\n'
Empty lines are tolerated when saving to text files.
>>> from tempfile import NamedTemporaryFile >>> tempFile2 = NamedTemporaryFile(delete=True) >>> tempFile2.close() >>> sc.parallelize(['', 'foo', '', 'bar', '']).saveAsTextFile(tempFile2.name) >>> ''.join(sorted(input(glob(tempFile2.name + "/part-0000*")))) '\\n\\n\\nbar\\nfoo\\n'
Using compressionCodecClass
>>> from tempfile import NamedTemporaryFile >>> tempFile3 = NamedTemporaryFile(delete=True) >>> tempFile3.close() >>> codec = "org.apache.hadoop.io.compress.GzipCodec" >>> sc.parallelize(['foo', 'bar']).saveAsTextFile(tempFile3.name, codec) >>> from fileinput import input, hook_compressed >>> result = sorted(input(glob(tempFile3.name + "/part*.gz"), openhook=hook_compressed)) >>> b''.join(result).decode('utf-8') 'bar\\nfoo\\n' """ else:
# Pair functions
""" Return the key-value pairs in this RDD to the master as a dictionary.
Notes ----- This method should only be used if the resulting data is expected to be small, as all the data is loaded into the driver's memory.
Examples -------- >>> m = sc.parallelize([(1, 2), (3, 4)]).collectAsMap() >>> m[1] 2 >>> m[3] 4 """
""" Return an RDD with the keys of each tuple.
Examples -------- >>> m = sc.parallelize([(1, 2), (3, 4)]).keys() >>> m.collect() [1, 3] """
""" Return an RDD with the values of each tuple.
Examples -------- >>> m = sc.parallelize([(1, 2), (3, 4)]).values() >>> m.collect() [2, 4] """
""" Merge the values for each key using an associative and commutative reduce function.
This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce.
Output will be partitioned with `numPartitions` partitions, or the default parallelism level if `numPartitions` is not specified. Default partitioner is hash-partition.
Examples -------- >>> from operator import add >>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)]) >>> sorted(rdd.reduceByKey(add).collect()) [('a', 2), ('b', 1)] """
""" Merge the values for each key using an associative and commutative reduce function, but return the results immediately to the master as a dictionary.
This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce.
Examples -------- >>> from operator import add >>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)]) >>> sorted(rdd.reduceByKeyLocally(add).items()) [('a', 2), ('b', 1)] """
""" Count the number of elements for each key, and return the result to the master as a dictionary.
Examples -------- >>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)]) >>> sorted(rdd.countByKey().items()) [('a', 2), ('b', 1)] """
""" Return an RDD containing all pairs of elements with matching keys in `self` and `other`.
Each pair of elements will be returned as a (k, (v1, v2)) tuple, where (k, v1) is in `self` and (k, v2) is in `other`.
Performs a hash join across the cluster.
Examples -------- >>> 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))] """
""" Perform a left outer join of `self` and `other`.
For each element (k, v) in `self`, the resulting RDD will either contain all pairs (k, (v, w)) for w in `other`, or the pair (k, (v, None)) if no elements in `other` have key k.
Hash-partitions the resulting RDD into the given number of partitions.
Examples -------- >>> x = sc.parallelize([("a", 1), ("b", 4)]) >>> y = sc.parallelize([("a", 2)]) >>> sorted(x.leftOuterJoin(y).collect()) [('a', (1, 2)), ('b', (4, None))] """
""" Perform a right outer join of `self` and `other`.
For each element (k, w) in `other`, the resulting RDD will either contain all pairs (k, (v, w)) for v in this, or the pair (k, (None, w)) if no elements in `self` have key k.
Hash-partitions the resulting RDD into the given number of partitions.
Examples -------- >>> x = sc.parallelize([("a", 1), ("b", 4)]) >>> y = sc.parallelize([("a", 2)]) >>> sorted(y.rightOuterJoin(x).collect()) [('a', (2, 1)), ('b', (None, 4))] """
""" Perform a right outer join of `self` and `other`.
For each element (k, v) in `self`, the resulting RDD will either contain all pairs (k, (v, w)) for w in `other`, or the pair (k, (v, None)) if no elements in `other` have key k.
Similarly, for each element (k, w) in `other`, the resulting RDD will either contain all pairs (k, (v, w)) for v in `self`, or the pair (k, (None, w)) if no elements in `self` have key k.
Hash-partitions the resulting RDD into the given number of partitions.
Examples -------- >>> 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))] """
# TODO: add option to control map-side combining # portable_hash is used as default, because builtin hash of None is different # cross machines. """ Return a copy of the RDD partitioned using the specified partitioner.
Examples -------- >>> pairs = sc.parallelize([1, 2, 3, 4, 2, 4, 1]).map(lambda x: (x, x)) >>> sets = pairs.partitionBy(2).glom().collect() >>> len(set(sets[0]).intersection(set(sets[1]))) 0 """
# Transferring O(n) objects to Java is too expensive. # Instead, we'll form the hash buckets in Python, # transferring O(numPartitions) objects to Java. # Each object is a (splitNumber, [objects]) pair. # In order to avoid too huge objects, the objects are # grouped into chunks.
# check used memory and avg size of chunk of objects or c > batch):
# let 1M < avg < 10M elif avg > 10: batch = max(int(batch / 1.5), 1)
keyed._jrdd.rdd()).asJavaPairRDD() id(partitionFunc))
# TODO: add control over map-side aggregation numPartitions=None, partitionFunc=portable_hash): """ Generic function to combine the elements for each key using a custom set of aggregation functions.
Turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a "combined type" C.
Users provide three functions:
- `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)
To avoid memory allocation, both mergeValue and mergeCombiners are allowed to modify and return their first argument instead of creating a new C.
In addition, users can control the partitioning of the output RDD.
Notes ----- V and C can be different -- for example, one might group an RDD of type (Int, Int) into an RDD of type (Int, List[Int]).
Examples -------- >>> 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])] """
partitionFunc=portable_hash): """ Aggregate the values of each key, using given combine functions and a neutral "zero value". This function can return a different result type, U, than the type of the values in this RDD, V. Thus, we need one operation for merging a V into a U and one operation for merging two U's, The former operation is used for merging values within a partition, and the latter is used for merging values between partitions. To avoid memory allocation, both of these functions are allowed to modify and return their first argument instead of creating a new U. """
lambda v: seqFunc(createZero(), v), seqFunc, combFunc, numPartitions, partitionFunc)
""" Merge the values for each key using an associative function "func" and a neutral "zeroValue" which may be added to the result an arbitrary number of times, and must not change the result (e.g., 0 for addition, or 1 for multiplication.).
Examples -------- >>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)]) >>> from operator import add >>> sorted(rdd.foldByKey(0, add).collect()) [('a', 2), ('b', 1)] """
partitionFunc)
# TODO: support variant with custom partitioner """ Group the values for each key in the RDD into a single sequence. Hash-partitions the resulting RDD with numPartitions partitions.
Notes ----- If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using reduceByKey or aggregateByKey will provide much better performance.
Examples -------- >>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)]) >>> sorted(rdd.groupByKey().mapValues(len).collect()) [('a', 2), ('b', 1)] >>> sorted(rdd.groupByKey().mapValues(list).collect()) [('a', [1, 1]), ('b', [1])] """
""" Pass each value in the key-value pair RDD through a flatMap function without changing the keys; this also retains the original RDD's partitioning.
Examples -------- >>> 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')] """
""" Pass each value in the key-value pair RDD through a map function without changing the keys; this also retains the original RDD's partitioning.
Examples -------- >>> x = sc.parallelize([("a", ["apple", "banana", "lemon"]), ("b", ["grapes"])]) >>> def f(x): return len(x) >>> x.mapValues(f).collect() [('a', 3), ('b', 1)] """
""" Alias for cogroup but with support for multiple RDDs.
Examples -------- >>> w = sc.parallelize([("a", 5), ("b", 6)]) >>> x = sc.parallelize([("a", 1), ("b", 4)]) >>> y = sc.parallelize([("a", 2)]) >>> z = sc.parallelize([("b", 42)]) >>> [(x, tuple(map(list, y))) for x, y in sorted(list(w.groupWith(x, y, z).collect()))] [('a', ([5], [1], [2], [])), ('b', ([6], [4], [], [42]))]
"""
# TODO: add variant with custom partitioner """ For each key k in `self` or `other`, return a resulting RDD that contains a tuple with the list of values for that key in `self` as well as `other`.
Examples -------- >>> 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], []))] """
""" Return a subset of this RDD sampled by key (via stratified sampling). Create a sample of this RDD using variable sampling rates for different keys as specified by fractions, a key to sampling rate map.
Examples -------- >>> fractions = {"a": 0.2, "b": 0.1} >>> rdd = sc.parallelize(fractions.keys()).cartesian(sc.parallelize(range(0, 1000))) >>> sample = dict(rdd.sampleByKey(False, fractions, 2).groupByKey().collect()) >>> 100 < len(sample["a"]) < 300 and 50 < len(sample["b"]) < 150 True >>> max(sample["a"]) <= 999 and min(sample["a"]) >= 0 True >>> max(sample["b"]) <= 999 and min(sample["b"]) >= 0 True """ RDDStratifiedSampler(withReplacement, fractions, seed).func, True)
""" Return each (key, value) pair in `self` that has no pair with matching key in `other`.
Examples -------- >>> x = sc.parallelize([("a", 1), ("b", 4), ("b", 5), ("a", 2)]) >>> y = sc.parallelize([("a", 3), ("c", None)]) >>> sorted(x.subtractByKey(y).collect()) [('b', 4), ('b', 5)] """
""" Return each value in `self` that is not contained in `other`.
Examples -------- >>> 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)] """ # note: here 'True' is just a placeholder
""" Creates tuples of the elements in this RDD by applying `f`.
Examples -------- >>> x = sc.parallelize(range(0,3)).keyBy(lambda x: x*x) >>> y = sc.parallelize(zip(range(0,5), range(0,5))) >>> [(x, list(map(list, y))) for x, y in sorted(x.cogroup(y).collect())] [(0, [[0], [0]]), (1, [[1], [1]]), (2, [[], [2]]), (3, [[], [3]]), (4, [[2], [4]])] """
""" Return a new RDD that has exactly numPartitions partitions.
Can increase or decrease the level of parallelism in this RDD. Internally, this uses a shuffle to redistribute data. If you are decreasing the number of partitions in this RDD, consider using `coalesce`, which can avoid performing a shuffle.
Examples -------- >>> rdd = sc.parallelize([1,2,3,4,5,6,7], 4) >>> sorted(rdd.glom().collect()) [[1], [2, 3], [4, 5], [6, 7]] >>> len(rdd.repartition(2).glom().collect()) 2 >>> len(rdd.repartition(10).glom().collect()) 10 """
""" Return a new RDD that is reduced into `numPartitions` partitions.
Examples -------- >>> 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]] """ # Decrease the batch size in order to distribute evenly the elements across output # partitions. Otherwise, repartition will possibly produce highly skewed partitions. else:
""" Zips this RDD with another one, returning key-value pairs with the first element in each RDD second element in each RDD, etc. Assumes that the two RDDs have the same number of partitions and the same number of elements in each partition (e.g. one was made through a map on the other).
Examples -------- >>> 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)] """
# use the smallest batchSize for both of them # auto batched or unlimited
# There will be an Exception in JVM if there are different number # of items in each partitions. other._jrdd_deserializer)
""" Zips this RDD with its element indices.
The ordering is first based on the partition index and then the ordering of items within each partition. So the first item in the first partition gets index 0, and the last item in the last partition receives the largest index.
This method needs to trigger a spark job when this RDD contains more than one partitions.
Examples -------- >>> sc.parallelize(["a", "b", "c", "d"], 3).zipWithIndex().collect() [('a', 0), ('b', 1), ('c', 2), ('d', 3)] """
""" Zips this RDD with generated unique Long ids.
Items in the kth partition will get ids k, n+k, 2*n+k, ..., where n is the number of partitions. So there may exist gaps, but this method won't trigger a spark job, which is different from :meth:`zipWithIndex`.
Examples -------- >>> sc.parallelize(["a", "b", "c", "d", "e"], 3).zipWithUniqueId().collect() [('a', 0), ('b', 1), ('c', 4), ('d', 2), ('e', 5)] """
""" Return the name of this RDD. """
""" Assign a name to this RDD.
Examples -------- >>> rdd1 = sc.parallelize([1, 2]) >>> rdd1.setName('RDD1').name() 'RDD1' """
""" A description of this RDD and its recursive dependencies for debugging. """ debug_string = self._jrdd.toDebugString() if debug_string: return debug_string.encode('utf-8')
""" Get the RDD's current storage level.
Examples -------- >>> rdd1 = sc.parallelize([1,2]) >>> rdd1.getStorageLevel() StorageLevel(False, False, False, False, 1) >>> print(rdd1.getStorageLevel()) Serialized 1x Replicated """ java_storage_level.useMemory(), java_storage_level.useOffHeap(), java_storage_level.deserialized(), java_storage_level.replication())
""" Returns the default number of partitions to use during reduce tasks (e.g., groupBy). If spark.default.parallelism is set, then we'll use the value from SparkContext defaultParallelism, otherwise we'll use the number of partitions in this RDD.
This mirrors the behavior of the Scala Partitioner#defaultPartitioner, intended to reduce the likelihood of OOMs. Once PySpark adopts Partitioner-based APIs, this behavior will be inherent. """ return self.ctx.defaultParallelism else:
""" Return the list of values in the RDD for key `key`. This operation is done efficiently if the RDD has a known partitioner by only searching the partition that the key maps to.
Examples -------- >>> l = range(1000) >>> rdd = sc.parallelize(zip(l, l), 10) >>> rdd.lookup(42) # slow [42] >>> sorted = rdd.sortByKey() >>> sorted.lookup(42) # fast [42] >>> sorted.lookup(1024) [] >>> rdd2 = sc.parallelize([(('a', 'b'), 'c')]).groupByKey() >>> list(rdd2.lookup(('a', 'b'))[0]) ['c'] """
""" Return a JavaRDD of Object by unpickling
It will convert each Python object into Java object by Pyrolite, whenever the RDD is serialized in batch or not. """
""" Approximate version of count() that returns a potentially incomplete result within a timeout, even if not all tasks have finished.
Examples -------- >>> rdd = sc.parallelize(range(1000), 10) >>> rdd.countApprox(1000, 1.0) 1000 """
""" Approximate operation to return the sum within a timeout or meet the confidence.
Examples -------- >>> rdd = sc.parallelize(range(1000), 10) >>> r = sum(range(1000)) >>> abs(rdd.sumApprox(1000) - r) / r < 0.05 True """
""" Approximate operation to return the mean within a timeout or meet the confidence.
Examples -------- >>> rdd = sc.parallelize(range(1000), 10) >>> r = sum(range(1000)) / 1000.0 >>> abs(rdd.meanApprox(1000) - r) / r < 0.05 True """
""" Return approximate number of distinct elements in the RDD.
Parameters ---------- relativeSD : float, optional Relative accuracy. Smaller values create counters that require more space. It must be greater than 0.000017.
Notes ----- The algorithm used is based on streamlib's implementation of `"HyperLogLog in Practice: Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available here <https://doi.org/10.1145/2452376.2452456>`_.
Examples -------- >>> n = sc.parallelize(range(1000)).map(str).countApproxDistinct() >>> 900 < n < 1100 True >>> n = sc.parallelize([i % 20 for i in range(1000)]).countApproxDistinct() >>> 16 < n < 24 True """ # the hash space in Java is 2^32
""" Return an iterator that contains all of the elements in this RDD. The iterator will consume as much memory as the largest partition in this RDD. With prefetch it may consume up to the memory of the 2 largest partitions.
Parameters ---------- prefetchPartitions : bool, optional If Spark should pre-fetch the next partition before it is needed.
Examples -------- >>> rdd = sc.parallelize(range(10)) >>> [x for x in rdd.toLocalIterator()] [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] """ self._jrdd.rdd(), prefetchPartitions)
""" Marks the current stage as a barrier stage, where Spark must launch all tasks together. In case of a task failure, instead of only restarting the failed task, Spark will abort the entire stage and relaunch all tasks for this stage. The barrier execution mode feature is experimental and it only handles limited scenarios. Please read the linked SPIP and design docs to understand the limitations and future plans.
.. versionadded:: 2.4.0
Returns ------- :class:`RDDBarrier` instance that provides actions within a barrier stage.
See Also -------- pyspark.BarrierTaskContext
Notes ----- For additional information see
- `SPIP: Barrier Execution Mode <http://jira.apache.org/jira/browse/SPARK-24374>`_ - `Design Doc <https://jira.apache.org/jira/browse/SPARK-24582>`_
This API is experimental """
""" Whether this RDD is in a barrier stage. """
""" Specify a :class:`pyspark.resource.ResourceProfile` to use when calculating this RDD. This is only supported on certain cluster managers and currently requires dynamic allocation to be enabled. It will result in new executors with the resources specified being acquired to calculate the RDD.
.. versionadded:: 3.1.0
Notes ----- This API is experimental """ else:
""" Get the :class:`pyspark.resource.ResourceProfile` specified with this RDD or None if it wasn't specified.
.. versionadded:: 3.1.0
Returns ------- :py:class:`pyspark.resource.ResourceProfile` The the user specified profile or None if none were specified
Notes ----- This API is experimental """ else:
# the serialized command will be compressed by broadcast # The broadcast will have same life cycle as created PythonRDD
sc.pythonVer, broadcast_vars, sc._javaAccumulator)
""" Wraps an RDD in a barrier stage, which forces Spark to launch tasks of this stage together. :class:`RDDBarrier` instances are created by :func:`RDD.barrier`.
.. versionadded:: 2.4.0
Notes ----- This API is experimental """
""" Returns a new RDD by applying a function to each partition of the wrapped RDD, where tasks are launched together in a barrier stage. The interface is the same as :func:`RDD.mapPartitions`. Please see the API doc there.
.. versionadded:: 2.4.0
Notes ----- This API is experimental """
""" Returns a new RDD by applying a function to each partition of the wrapped RDD, while tracking the index of the original partition. And all tasks are launched together in a barrier stage. The interface is the same as :func:`RDD.mapPartitionsWithIndex`. Please see the API doc there.
.. versionadded:: 3.0.0
Notes ----- This API is experimental """
""" Examples -------- Pipelined maps:
>>> rdd = sc.parallelize([1, 2, 3, 4]) >>> rdd.map(lambda x: 2 * x).cache().map(lambda x: 2 * x).collect() [4, 8, 12, 16] >>> rdd.map(lambda x: 2 * x).map(lambda x: 2 * x).collect() [4, 8, 12, 16]
Pipelined reduces:
>>> from operator import add >>> rdd.map(lambda x: 2 * x).reduce(add) 20 >>> rdd.flatMap(lambda x: [x, x]).reduce(add) 20 """
# This transformation is the first in its stage: else:
prev.preservesPartitioning and preservesPartitioning
def _jrdd(self):
else:
self._jrdd_deserializer, profiler) self.preservesPartitioning, self.is_barrier)
# The small batch size here ensures that we see multiple batches, # even in these small test examples: globs=globs, optionflags=doctest.ELLIPSIS) sys.exit(-1)
|