Pyspark：洗牌 RDD

Question

我正在尝试随机化 RDD 中元素的顺序。我目前的方法是使用随机整数的 RDD 来压缩元素，然后通过这些整数加入。

然而，pyspark 只用了 100000000 个整数。我正在使用下面的代码。

我的问题是：是否有更好的方法来压缩随机索引或随机播放？

我试过按随机键排序，效果不错，但速度很慢。

def random_indices(n):
    """
    return an iterable of random indices in range(0,n)
    """
    indices = range(n)
    random.shuffle(indices)
    return indices

pyspark 中发生以下情况：

Using Python version 2.7.3 (default, Jun 22 2015 19:33:41)
SparkContext available as sc.
>>> import clean
>>> clean.sc = sc
>>> clean.random_indices(100000000)
Killed

Answer 1

一种可能的方法是使用 mapParitions

添加随机密钥

import os
import numpy as np

swap = lambda x: (x[1], x[0])

def add_random_key(it):
    # make sure we get a proper random seed
    seed = int(os.urandom(4).encode('hex'), 16) 
    # create separate generator
    rs = np.random.RandomState(seed)
    # Could be randint if you prefer integers
    return ((rs.rand(), swap(x)) for x in it)

rdd_with_keys = (rdd
  # It will be used as final key. If you don't accept gaps 
  # use zipWithIndex but this should be cheaper 
  .zipWithUniqueId()
  .mapPartitions(add_random_key, preservesPartitioning=True))

接下来您可以重新分区，对每个分区进行排序并提取值：

n = rdd.getNumPartitions()
(rdd_with_keys
    # partition by random key to put data on random partition 
    .partitionBy(n)
    # Sort partition by random value to ensure random order on partition
    .mapPartitions(sorted, preservesPartitioning=True)
    # Extract (unique_id, value) pairs
    .values())

如果按分区排序仍然很慢，可以用 Fisher–Yates shuffle 代替。

如果你只需要一个随机数据那么你可以使用mllib.RandomRDDs

from pyspark.mllib.random import RandomRDDs

RandomRDDs.uniformRDD(sc, n)

理论上它可以用输入压缩 rdd 但它需要匹配每个分区的元素数。

Answer 2

pyspark 成功了！

from random import randrange
data_rnd = data.sortBy(lambda x: randrange(1000000))

Pyspark：洗牌 RDD

Pyspark: shuffle RDD

python

hadoop

bigdata

apache-spark

pyspark