pyspark - как выбрать точное количество записей на страты, используя (df.sampleByKey ()) в стратифицированной случайной выборке - PullRequest
Новая
       8

pyspark - как выбрать точное количество записей на страты, используя (df.sampleByKey ()) в стратифицированной случайной выборке

0 голосов
/ 09 апреля 2020

У меня есть фрейм данных spark (я использую pyspark) 'orders' . В нем есть следующие столбцы 

['id', 'orderdate', 'customerid', 'status']

Я пытаюсь сделать стратифицированную случайную выборку, используя ключевой столбец в качестве «статуса». Моя цель, как показано ниже 

>> create a new dataframe with exactly 5 random records per status

Поэтому выбранный мною метод использует .sampleBy ('strata_key', {фракция_dict}). Но проблема, с которой я столкнулся, заключается в выборе точного значения дроби для каждого статуса, чтобы каждый раз я получал ровно 5 случайных записей на каждый статус. Я следовал ниже метод

1.Создал словарь для общего подсчета на статус, как показано ниже 

#Total count of records for each order 'status' in 'ORDERS' dataframe is as below

d=dict([(x['status'],x['count']) for x in orders.groupBy("status").count().collect()])
print(d)

ВЫХОД: 

{'PENDING_PAYMENT': 15030, 'COMPLETE': 22899, 'ON_HOLD': 3798, 'PAYMENT_REVIEW': 729, 'PROCESSING': 8275, 'CLOSED': 7556, 'SUSPECTED_FRAUD': 1558, 
'PENDING': 7610, 'CANCELED': 1428}

2.Создана функция, которая генерирует значения дроби, необходимые для выборки точных N записей 

#Exact number of records needed per status
N=5

#function calculates fraction

def fraction_calc(count_dict,N)
    d_mod={}
    for i in d:
        d_mod[i]=(N/d[i])
    return d_mod

#creating dictionary of fractions using above function
fraction=fraction_calc(d,5)
print(fraction)

ВЫХОД: 

{'PENDING_PAYMENT': 0.00033266799733865603, 'COMPLETE': 0.000218350146294598, 'ON_HOLD': 0.0013164823591363876, 'PAYMENT_REVIEW': 0.006858710562414266, 'PROCESSING': 0.0006042296072507553, 'CLOSED': 0.0006617257808364214, 'SUSPECTED_FRAUD': 0.003209242618741977, 'PENDING': 0.000657030223390276, 'CANCELED': 0.0035014005602240898}

3.Создание окончательного фрейма данных, который выбирается с использованием API для запуска с выборкой .sampleBy () 

#creating final sampled dataframe
df_sample=orders.sampleBy("status",fraction)

Но, тем не менее, я не получаю 5 точных записей на состояние. Пример вывода как показано ниже 

#Checking count per status of resultant sample dataframe
df_sample.groupBy("status").count().show()
+---------------+-----+
|         status|count|
+---------------+-----+
|PENDING_PAYMENT|    3|
|       COMPLETE|    6|
|        ON_HOLD|    7|
| PAYMENT_REVIEW|    4|
|     PROCESSING|    6|
|         CLOSED|    6|
|SUSPECTED_FRAUD|    7|
|        PENDING|    9|
|       CANCELED|    5|
+---------------+-----+

Что я должен сделать здесь, чтобы достичь своей цели.

1 Ответ

0 голосов
/ 11 апреля 2020

Нашел работу около 

from pyspark.sql.window import Window
from pyspark.sql.functions import rand,row_number

1. Использование встроенной функции rand () для генерации столбца «ключ» из случайных чисел и последующего присвоения номера строки каждому элементу окна раздела, созданного в порядке столбцов «order_status» с помощью «ключа». Код следующий: 

df_sample=df.withColumn("key",rand()).\
withColumn("rnk", row_number().\
over(Window.partitionBy("status").\
orderBy("key"))).\
where("rnk<=5").drop("key","rnk")

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

#Checking count per status of resultant sample dataframe
df_sample.groupBy("status").count().show()
+---------------+-----+
|   status      |count|
+---------------+-----+
|PENDING_PAYMENT|    5|
|       COMPLETE|    5|
|        ON_HOLD|    5|
| PAYMENT_REVIEW|    5|
|     PROCESSING|    5|
|         CLOSED|    5|
|SUSPECTED_FRAUD|    5|
|        PENDING|    5|
|       CANCELED|    5|
+---------------+-----+
...