python - 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 压缩，但它需要匹配每个分区的元素数量。