Cómo iterar sobre filas y columnas en el marco de datos de PySpark

En este artículo, discutiremos cómo iterar filas y columnas en el marco de datos de PySpark.

Cree el marco de datos para la demostración:

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
dataframe.show()

Producción:

Método 1: usar recopilar()

Este método recopilará todas las filas y columnas del marco de datos y luego lo recorrerá usando for loop. Aquí se usa un iterador para iterar sobre un ciclo de los elementos recolectados usando el método collect().

Sintaxis :

for itertator in dataframe.collect():
                    print(itertator["column_name"],...............)

dónde,

• dataframe es el dataframe de entrada
• itertator se utiliza para recopilar filas
• column_name es la columna para iterar filas

Ejemplo: Aquí vamos a iterar todas las columnas en el dataframe con el método collect() y dentro del ciclo for, estamos especificando iterator[‘column_name’] para obtener los valores de las columnas.

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
# using collect
for i in dataframe.collect():
    # display
    print(i["ID"], i["NAME"], i["Company"])

Producción:

Método 2: Usando toLocalIterator()

Devolverá el iterador que contiene todas las filas y columnas en RDD. Es similar al método collect(), pero está en formato rdd, por lo que está disponible dentro del método rdd. Podemos usar toLocalIterator() con rdd como:

dataframe.rdd.toLocalIterator()

Para iterar todas las filas y columnas, estamos iterando esto dentro de un bucle for

Sintaxis :

for itertator in dataframe.rdd.toLocalIterator():
                    print(itertator["column_name"],...............)

dónde,

• dataframe es el dataframe de entrada
• itertator se utiliza para recopilar filas
• column_name es la columna para iterar filas

Ejemplo : aquí vamos a iterar todas las columnas en el marco de datos con el método toLocalIterator() y dentro del ciclo for, estamos especificando iterator[‘column_name’] para obtener los valores de las columnas.

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
# using toLocalIterator()
for i in dataframe.rdd.toLocalIterator():
   
    # display
    print(i["ID"], i["NAME"], i["Company"])

Producción:

Método 3: Usar iterrows()

Esto iterará filas. Antes de eso, tenemos que convertir nuestro dataframe de PySpark en un dataframe de Pandas usando el método toPandas(). Este método se utiliza para iterar fila por fila en el marco de datos.

Sintaxis : dataframe.toPandas().iterrows()

Ejemplo: En este ejemplo, vamos a iterar filas de tres columnas usando iterrows() usando for loop.

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
# using iterrows()
for index, row in dataframe.toPandas().iterrows():
    # display with index
    print(row[0], row[1], row[2])

Producción:

Método 4: Usar select()

La función select() se utiliza para seleccionar el número de columnas. entonces estamos usando la función de recopilación() para obtener las filas a través del bucle for.

El método de selección seleccionará las columnas que se mencionan y obtendrá los datos de la fila utilizando el método de recopilación(). Este método recopilará filas de las columnas dadas.

Sintaxis: dataframe.select(“columna1″,…………,”columna n”).collect()

Ejemplo: Aquí vamos a seleccionar las columnas ID y Nombre del marco de datos dado usando el método select()

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
# select only id and company
for rows in dataframe.select("ID", "Name").collect():
       # display
    print(rows[0], rows[1])

Producción:

Método 5: Usar la comprensión de listas

Esto actuará como un ciclo para obtener cada fila y, finalmente, podemos usar el ciclo for para obtener columnas particulares. Vamos a iterar los datos en la columna dada usando el método collect() a través de rdd.

Sintaxis : dataframe.rdd.collect()

Ejemplo: aquí vamos a iterar filas en la columna NOMBRE.

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
# select name column
for i in [j["NAME"] for j in dataframe.rdd.collect()]:
    print(i)

Producción:

sravan
ojaswi
rohith
sridevi
bobby

Método 6: Usar mapa()

En este método, usaremos la función map(), que devuelve una nueva v de un marco de datos o RDD dado. La función map() se usa con la función lambda para iterar a través de cada fila del marco de datos pyspark.

Para recorrer cada fila usando map() primero, tenemos que convertir el marco de datos de PySpark en RDD porque map() se realiza solo en RDD, así que primero conviértalo en RDD y luego use map() en el cual, función lambda para iterar a través de cada fila y almacena el nuevo RDD en alguna variable y luego vuelve a convertir ese nuevo RDD en Dataframe usando toDF() pasándole el esquema.

Sintaxis:

rdd=dataframe.rdd.map(lambda loop: (
      loop["column1"],...,loop["columnn"]) )
 rdd.toDF(["column1",.......,"columnn"]).collect()

Ejemplo: aquí vamos a iterar la columna ID y NOMBRE

# importing module
import pyspark
 
# importing sparksession from pyspark.sql module
from pyspark.sql import SparkSession
 
# creating sparksession and giving an app name
spark = SparkSession.builder.appName('sparkdf').getOrCreate()
 
# list  of employee data
data = [["1", "sravan", "company 1"],
        ["2", "ojaswi", "company 1"],
        ["3", "rohith", "company 2"],
        ["4", "sridevi", "company 1"],
        ["5", "bobby", "company 1"]]
 
# specify column names
columns = ['ID', 'NAME', 'Company']
 
# creating a dataframe from the lists of data
dataframe = spark.createDataFrame(data, columns)
 
# select id and name column using map()
rdd = dataframe.rdd.map(lambda loop: (
    loop["ID"], loop["NAME"]))
 
# convert to dataframe and display
rdd.toDF(["ID", "NAME"]).collect()

Producción:

[Row(ID='1', NAME='sravan'),
Row(ID='2', NAME='ojaswi'),
Row(ID='3', NAME='rohith'),
Row(ID='4', NAME='sridevi'),
Row(ID='5', NAME='bobby')]