¿Cómo duplicar una fila N en el marco de datos de Pyspark?

En este artículo, vamos a aprender cómo duplicar una fila N veces en un PySpark DataFrame.

Método 1: Repetir filas según el valor de la columna

En este método, primero crearemos un PySpark DataFrame usando createDataFrame() . En nuestro ejemplo, la columna «Y» tiene un valor numérico que solo se puede usar aquí para repetir filas. Usaremos la función withColumn() aquí y su parámetro expr se explicará a continuación.

Sintaxis :

DataFrame.withColumn(colName,col)

Parámetros : 

• colName : str nombre de la nueva columna
• col : Column(DataType) una expresión de columna de la nueva columna

El colName aquí es «Y». La expresión col que usaremos aquí es:

explode(array_repeat(Y,int(Y)))

• array_repeat es una expresión que crea una array que contiene una columna que cuenta veces repetidas.
• explotar es una expresión que devuelve una nueva fila para cada elemento en la array o mapa dado.

Ejemplo:

# Importing PySpark and Pandas
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col,expr
 
# Session Creation
Spark_Session = SparkSession.builder.appName(
    'Spark Session'
).getOrCreate()
 
# Accepting n from the user.
n = int(input('Enter n : '))
 
# Data filled in our DataFrame
rows = [['a',1,'@'],
        ['b',3,'_'],
        ['c',2,'!'],
        ['d',6,'(']]
 
# Columns of our DataFrame
columns = ['X','Y','Z']
 
# DataFrame is created
df = Spark_Session.createDataFrame(rows,columns)
 
# Printing the DataFrame
df.show()
 
# Creating a new DataFrame with a
# expression using functions
new_df = df.withColumn(
  "Y", expr("explode(array_repeat(Y,int(Y)))"))
 
# Printing the new DataFrame
new_df.show()

Producción : 

Método 2: usar recopilar() y agregar una fila aleatoria en la lista

En este método, primero aceptaremos N del usuario. Luego crearemos un PySpark DataFrame usando createDataFrame() . Luego podemos almacenar la lista de objetos Row encontrados usando el método collect() . La sintaxis necesaria es: 

DataFrame.collect()

en una variable. Luego usaremos la función Python List append() para agregar un objeto de fila en la lista que se realizará en un ciclo de N iteraciones. Finalmente, la lista de objetos Row se convertirá en un PySpark DataFrame.

Ejemplo:

# Importing PySpark and random
import pyspark
from pyspark.sql import SparkSession
import random
 
# Session Creation
Spark_Session = SparkSession.builder.appName(
    'Spark Session'
).getOrCreate()
 
# Accepting n from the user.
n = int(input('Enter n : '))
 
# Data filled in our DataFrame
rows = [['a',1,'@'],
        ['b',3,'_'],
        ['c',2,'!'],
        ['d',6,'(']]
 
# Columns of our DataFrame
columns = ['X','Y','Z']
 
# DataFrame is created
df = Spark_Session.createDataFrame(rows,columns)
 
# Showing the DataFrame
df.show()
 
# Creating a list of rows and
# getting a random row from the list
row_list = df.collect()
repeated = random.choice(row_list)
 
# adding a row object to the list
# n times
for _ in range(n):
  row_list.append(repeated)
 
# Final DataFrame
df = Spark_Session.createDataFrame(row_list)
 
# Result
df.show()

Producción : 

Método 3: convertir PySpark DataFrame en Pandas DataFrame

En este método, primero aceptaremos N del usuario. Luego crearemos un PySpark DataFrame usando createDataFrame() . Luego convertiremos un PySpark DataFrame en un Pandas DataFrame usando toPandas() . Luego obtendremos la primera fila del DataFrame usando el corte con Syntax DataFrame[:1]. Luego usaremos la función append() para pegar la fila al Pandas DataFrame usando un bucle. La sintaxis de append() es: 

Sintaxis : DataFrame.append(otro, ignore_index=Falso, verificar_integridad=Falso, ordenar=Falso)

Parámetros : 

• otro : DataFrame/Numpy Series Los datos que se agregarán
• ignore_index : bool, default: False Compruebe si el DataFrame del nuevo DataFrame depende del DataFrame anterior
• verificar_integridad : booleano, predeterminado: falso Se encarga de los valores duplicados
• sort : bool, default : False Ordenar columnas según el valor

Ejemplo:

# Importing PySpark and Pandas
import pyspark
from pyspark.sql import SparkSession
import pandas as pd
 
# Session Creation
Spark_Session = SparkSession.builder.appName(
    'Spark Session'
).getOrCreate()
 
# Accepting n from the user.
n = int(input('Enter n : '))
 
# Data filled in our DataFrame
rows = [['a',1,'@'],
        ['b',3,'_'],
        ['c',2,'!'],
        ['d',6,'(']]
 
# Columns of our DataFrame
columns = ['X','Y','Z']
 
# DataFrame is created
df = Spark_Session.createDataFrame(rows,columns)
 
# Converting to a Pandas DataFrame
df_pandas = df.toPandas()
 
# The initial DataFrame
print('First DF')
print(df_pandas)
 
# the first row
first_row = df_pandas[:1]
 
# Appending the row n times
for _ in range(n):
  df_pandas = df_pandas.append(first_row,ignore_index = True)
 
# Final DataFrame
print('New DF')
print(df_pandas)

Producción :