Итерация по строкам фрейма данных pyspark, но каждая строка хранится как фрейм данных - PullRequest
Новая
       9

Итерация по строкам фрейма данных pyspark, но каждая строка хранится как фрейм данных

1 голос
/ 31 марта 2020

Я новичок в Pyspark. Здесь я хочу l oop через каждую строку DataFrame. Я хочу, чтобы каждая строка снова была в виде фрейма данных.

Пример:

Это мой первый фрейм данных 

+----------+---------------------+-------+
|account_id|transaction_timestamp|balance|
+----------+---------------------+-------+
|      8881|        1582047353000| 500.00|
|        45|        1582198671000| 500.00|
|         3|        1582047353000| 500.00|
+----------+---------------------+-------+

Теперь я хочу вывод, как показано ниже: 

+----------+---------------------+-------+
|account_id|transaction_timestamp|balance|
+----------+---------------------+-------+
|      8881|        1582047353000| 500.00|
+----------+---------------------+-------+

+----------+---------------------+-------+
|account_id|transaction_timestamp|balance|
+----------+---------------------+-------+
|        45|        1582198671000| 500.00|
+----------+---------------------+-------+

+----------+---------------------+-------+
|account_id|transaction_timestamp|balance|
+----------+---------------------+-------+
|         3|        1582047353000| 500.00|
+----------+---------------------+-------+

1 Ответ

1 голос
/ 31 марта 2020

Мы можем добавить row_number к каждой строке, а затем повторять по каждой строке, мы можем разделить фрейм данных на основе значения row_number.

Example: 

df.show()
#+----------+---------------------+-------+
#|account_id|transaction_timestamp|balance|
#+----------+---------------------+-------+
#|      8881|        1582047353000| 500.00|
#|        45|        1582047353000| 500.00|
#|         3|        1582047353000| 500.00|
#+----------+---------------------+-------+
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window

df1=df.withColumn("rn",row_number().over(w)).show()

dict_df = {}

#get max row_number value
max_val=df1.select(max("rn")).collect()[0][0]

#loop until max value and filter based on row_number and add into dict_df dictionary.
for x in range(max_val):
    dict_df[x+1] = df1.filter(col('rn')==x+1).drop(col('rn'))

dict_df
#{1: DataFrame[account_id: string, transaction_timestamp: string, balance: string], 2: DataFrame[account_id: string, transaction_timestamp: string, balance: string], 3: DataFrame[account_id: string, transaction_timestamp: string, balance: string]}

#accessing dict_df

dict_df[1].show()
#+----------+---------------------+-------+
#|account_id|transaction_timestamp|balance|
#+----------+---------------------+-------+
#|      8881|        1582047353000| 500.00|
#+----------+---------------------+-------+

dict_df[2].show()
#+----------+---------------------+-------+
#|account_id|transaction_timestamp|balance|
#+----------+---------------------+-------+
#|        45|        1582047353000| 500.00|
#+----------+---------------------+-------+

dict_df[3].show()
#+----------+---------------------+-------+
#|account_id|transaction_timestamp|balance|
#+----------+---------------------+-------+
#|         3|        1582047353000| 500.00|
#+----------+---------------------+-------+

#using globals() to define variable(not recommended)
for x in range(max_val):
    globals()['df_{}'.format(x+1)] = df1.filter(col('rn')==x+1).drop(col('rn'))

df_1.show()
#+----------+---------------------+-------+
#|account_id|transaction_timestamp|balance|
#+----------+---------------------+-------+
#|      8881|        1582047353000| 500.00|
#+----------+---------------------+-------+
...