如何从PySpark DataFrame中随机抽取一行

Question

如何从PySpark DataFrame中获取随机行？我只看到了以分数为参数的方法sample()。将这个分数设置为1/numberOfRows会导致随机结果，有时我不会得到任何行。

在RDD上有一个方法takeSample()，它将您希望样本包含的元素数量作为参数。我知道这可能很慢，因为你必须计算每个分区，但有没有办法在DataFrame上获得这样的东西？

Answer 1

您可以简单地在RDD:上调用takeSample

df = sqlContext.createDataFrame(
    [(1, "a"), (2, "b"), (3, "c"), (4, "d")], ("k", "v"))
df.rdd.takeSample(False, 1, seed=0)
## [Row(k=3, v='c')]

如果你不想收集，你可以简单地采取更高的分数和限制：

df.sample(False, 0.1, seed=0).limit(1)

---

不要传递seed，每次都应该获得不同的DataFrame。

Answer 2

不同类型的样品

随机抽样%的数据，包括和不包括替换

import pyspark.sql.functions as F
#Randomly sample 50% of the data without replacement
sample1 = df.sample(False, 0.5, seed=0)
#Randomly sample 50% of the data with replacement
sample1 = df.sample(True, 0.5, seed=0)
#Take another sample exlcuding records from previous sample using Anti Join
sample2 = df.join(sample1, on='ID', how='left_anti').sample(False, 0.5, seed=0)
#Take another sample exlcuding records from previous sample using Where
sample1_ids = [row['ID'] for row in sample1.ID]
sample2 = df.where(~F.col('ID').isin(sample1_ids)).sample(False, 0.5, seed=0)
#Generate a startfied sample of the data across column(s)
#Sampling is probabilistic and thus cannot guarantee an exact number of rows
fractions = {
        'NJ': 0.5, #Take about 50% of records where state = NJ
    'NY': 0.25, #Take about 25% of records where state = NY
    'VA': 0.1, #Take about 10% of records where state = VA
}
stratified_sample = df.sampleBy(F.col('state'), fractions, seed=0)

Answer 3

这里有一个使用Pandas DataFrame.Sample方法的替代方法。这使用spark applyInPandas方法来分配组，可从spark 3.0.0获得。这允许您选择每个组的确切行数。

我已经将args和kwargs添加到函数中，这样您就可以访问DataFrame.Sample的其他参数。

def sample_n_per_group(n, *args, **kwargs):
    def sample_per_group(pdf):
        return pdf.sample(n, *args, **kwargs)
    return sample_per_group
df = spark.createDataFrame(
    [
        (1, 1.0), 
        (1, 2.0), 
        (2, 3.0), 
        (2, 5.0), 
        (2, 10.0)
    ],
    ("id", "v")
)
(df.groupBy("id")
   .applyInPandas(
        sample_n_per_group(1, random_state=2), 
        schema=df.schema
   )
)

从文件中了解大型团体的局限性：

此功能需要完全洗牌。一个组的所有数据加载到内存中，因此用户应该意识到潜在的OOM如果数据有偏差，并且某些群体太大而无法容纳，则存在风险记忆力