Генерация случайных чисел в PySpark

Давайте начнем с простой функции, которая всегда возвращает случайное целое число:

import numpy as np

def f(x):
    return np.random.randint(1000)

и RDD, заполненный нулями и отображаемый с использованием f:

rdd = sc.parallelize([0] * 10).map(f)

Поскольку выше RDD не сохраняется, я ожидаю, что каждый раз, когда я соберусь, получаю другой вывод:

> rdd.collect()
[255, 512, 512, 512, 255, 512, 255, 512, 512, 255]

Если мы игнорируем тот факт, что распределение значений на самом деле не выглядит случайным, это более или менее то, что происходит. Проблема начинается, когда мы берем только первый элемент:

assert len(set(rdd.first() for _ in xrange(100))) == 1

или

assert len(set(tuple(rdd.take(1)) for _ in xrange(100))) == 1

Кажется, каждый раз возвращает одинаковое число. Я смог воспроизвести это поведение на двух разных машинах с Spark 1.2, 1.3 и 1.4. Здесь я использую np.random.randint, но он ведет себя одинаково с random.randint.

Эта проблема, такая же, как неточно-случайные результаты с collect, кажется специфичной для Python, и я не смог воспроизвести ее с помощью Scala:

def f(x: Int) = scala.util.Random.nextInt(1000)

val rdd = sc.parallelize(List.fill(10)(0)).map(f)
(1 to 100).map(x => rdd.first).toSet.size

rdd.collect()

Я пропустил что-то очевидное здесь?

Edit

Оказывается, источником проблемы является реализация RNG Python. Чтобы процитировать официальную документацию:

Функции, предоставляемые этим модулем, являются фактически связанными методами скрытого экземпляра класса random.Random. Вы можете создавать собственные экземпляры Random для генерации генераторов, не имеющих общего состояния.

Я предполагаю, что NumPy работает так же и переписывает f с помощью экземпляра RandomState следующим образом

import os
import binascii

def f(x, seed=None):
    seed = (
        seed if seed is not None 
        else int(binascii.hexlify(os.urandom(4)), 16))
    rs = np.random.RandomState(seed)
    return rs.randint(1000)

делает это медленнее, но решает проблему.

В то время как выше объясняется не случайные результаты сбора, я до сих пор не понимаю, как это влияет на first/take(1) между несколькими действиями.

Ответ 1

Таким образом, фактическая проблема здесь относительно проста. Каждый подпроцесс в Python наследует свое состояние от родителя:

len(set(sc.parallelize(range(4), 4).map(lambda _: random.getstate()).collect()))
# 1

Поскольку родительское состояние не имеет причин для изменения в этом конкретном сценарии, а работники имеют ограниченный срок службы, состояние каждого ребенка будет одинаковым для каждого прогона.

Ответ 2

Кажется, это ошибка (или функция) randint. Я вижу такое же поведение, но как только я изменяю f, значения действительно меняются. Итак, я не уверен в фактической случайности этого метода.... Я не могу найти какую-либо документацию, но, похоже, использует какой-то детерминированный математический алгоритм вместо использования более переменных функций работающей машины. Даже если я иду туда и обратно, цифры, похоже, совпадают при возвращении к исходному значению...

Ответ 3

Для моего варианта использования большая часть решения была похоронена в редактировании в нижней части вопроса. Однако есть еще одна сложность: я хотел использовать одну и ту же функцию для генерации нескольких (разных) случайных столбцов. Оказывается, у Spark есть предположение, что выход UDF является детерминированным, что означает, что он может пропустить последующие вызовы той же функции с теми же входами. Для функций, которые возвращают случайный вывод, это явно не то, что вам нужно.

Чтобы обойти это, я сгенерировал отдельный начальный столбец для каждого случайного столбца, который хотел, используя встроенную функцию PySpark rand:

import pyspark.sql.functions as F
from pyspark.sql.types import IntegerType
import numpy as np

@F.udf(IntegerType())
def my_rand(seed):
    rs = np.random.RandomState(seed)
    return rs.randint(1000)

seed_expr = (F.rand()*F.lit(4294967295).astype('double')).astype('bigint')
my_df = (
    my_df
    .withColumn('seed_0', seed_expr)
    .withColumn('seed_1', seed_expr)
    .withColumn('myrand_0', my_rand(F.col('seed_0')))
    .withColumn('myrand_1', my_rand(F.col('seed_1')))
    .drop('seed_0', 'seed_1')
)

Я использую API DataFrame, а не RDD исходной проблемы, потому что это то, с чем я более знаком, но предположительно применяются те же концепции.

NB. По-видимому, можно отключить допущение детерминизма для пользовательских функций Scala Spark начиная с версии 2.3: https://jira.apache.org/jira/browse/SPARK-20586.