python/pyspark/rdd.py
def partitionBy(self, numPartitions, partitionFunc=hash):
        """
        Return a copy of the RDD partitioned using the specified partitioner.

        >>> pairs = sc.parallelize([1, 2, 3, 4, 2, 4, 1]).map(lambda x: (x, x))
        >>> sets = pairs.partitionBy(2).glom().collect()
        >>> set(sets[0]).intersection(set(sets[1]))
        set([])
        """
        if numPartitions is None:
            numPartitions = self.ctx.defaultParallelism
        # 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.
        outputSerializer = self.ctx._unbatched_serializer
        def add_shuffle_key(split, iterator):

            buckets = defaultdict(list)

            for (k, v) in iterator:
                buckets[partitionFunc(k) % numPartitions].append((k, v))
            for (split, items) in buckets.iteritems():
                yield pack_long(split)
                yield outputSerializer.dumps(items)
        keyed = PipelinedRDD(self, add_shuffle_key)
        keyed._bypass_serializer = True
        with _JavaStackTrace(self.context) as st:
            pairRDD = self.ctx._jvm.PairwiseRDD(keyed._jrdd.rdd()).asJavaPairRDD()
            partitioner = self.ctx._jvm.PythonPartitioner(numPartitions,
                                                          id(partitionFunc))
        jrdd = pairRDD.partitionBy(partitioner).values()
        rdd = RDD(jrdd, self.ctx, BatchedSerializer(outputSerializer))
        # This is required so that id(partitionFunc) remains unique, even if
        # partitionFunc is a lambda:
        rdd._partitionFunc = partitionFunc
        return rdd

        Return a copy of the RDD partitioned using the specified partitioner.

        >>> pairs = sc.parallelize([1, 2, 3, 4, 2, 4, 1]).map(lambda x: (x, x))
        >>> sets = pairs.partitionBy(2).glom().collect()
        >>> set(sets[0]).intersection(set(sets[1]))
        set([])
        
Return a copy of the RDD partitioned using the specified partitioner.
Return a copy of the RDD partitioned using the specified partitioner .
def partitionBy(self, numPartitions, partitionFunc=hash):
        if numPartitions is None:
            numPartitions = self.ctx.defaultParallelism
        outputSerializer = self.ctx._unbatched_serializer
        def add_shuffle_key(split, iterator):
            buckets = defaultdict(list)
            for (k, v) in iterator:
                buckets[partitionFunc(k) % numPartitions].append((k, v))
            for (split, items) in buckets.iteritems():
                yield pack_long(split)
                yield outputSerializer.dumps(items)
        keyed = PipelinedRDD(self, add_shuffle_key)
        keyed._bypass_serializer = True
        with _JavaStackTrace(self.context) as st:
            pairRDD = self.ctx._jvm.PairwiseRDD(keyed._jrdd.rdd()).asJavaPairRDD()
            partitioner = self.ctx._jvm.PythonPartitioner(numPartitions,
                                                          id(partitionFunc))
        jrdd = pairRDD.partitionBy(partitioner).values()
        rdd = RDD(jrdd, self.ctx, BatchedSerializer(outputSerializer))
        rdd._partitionFunc = partitionFunc
        return rdd
def partitionBy ( self , numPartitions , partitionFunc = hash ) : if numPartitions is None : numPartitions = self . ctx . defaultParallelism outputSerializer = self . ctx . _unbatched_serializer def add_shuffle_key ( split , iterator ) : buckets = defaultdict ( list ) for ( k , v ) in iterator : buckets [ partitionFunc ( k ) % numPartitions ] . append ( ( k , v ) ) for ( split , items ) in buckets . iteritems ( ) : yield pack_long ( split ) yield outputSerializer . dumps ( items ) keyed = PipelinedRDD ( self , add_shuffle_key ) keyed . _bypass_serializer = True with _JavaStackTrace ( self . context ) as st : pairRDD = self . ctx . _jvm . PairwiseRDD ( keyed . _jrdd . rdd ( ) ) . asJavaPairRDD ( ) partitioner = self . ctx . _jvm . PythonPartitioner ( numPartitions , id ( partitionFunc ) ) jrdd = pairRDD . partitionBy ( partitioner ) . values ( ) rdd = RDD ( jrdd , self . ctx , BatchedSerializer ( outputSerializer ) ) rdd . _partitionFunc = partitionFunc return rdd
Return a copy of the RDD partitioned using the specified partitioner.
Return a copy of the RDD partitioned using the specified partitioner .
"""
        Return a copy of the RDD partitioned using the specified partitioner.

        >>> pairs = sc.parallelize([1, 2, 3, 4, 2, 4, 1]).map(lambda x: (x, x))
        >>> sets = pairs.partitionBy(2).glom().collect()
        >>> set(sets[0]).intersection(set(sets[1]))
        set([])
        """ # 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. # This is required so that id(partitionFunc) remains unique, even if # partitionFunc is a lambda:
self
numPartitions
partitionFunc