Convertir pyspark rdd a dataFrame
RDD significa conjuntos de datos distribuidos resilientes. Podemos llamar a RDD una estructura de datos fundamental en Apache Spark.
Sintaxis
Spark_app.sparkcontext.Paralelizar (datos)Podemos relacionar los datos en un formato tabular. La estructura de datos utilizada es DataFrame.El formato tabular significa que almacena datos en filas y columnas.
En Pyspark, podemos crear un marcado de datos desde la aplicación Spark con el método createATATAFRame ().
Sintaxis
Spark_app.creatataFrame (input_data, columnas)Donde input_data puede ser un diccionario o una lista para crear un marco de datos a partir de estos datos, y si el input_data es una lista de diccionarios, entonces las columnas no son necesarias. Si es una lista anidada, tenemos que proporcionar los nombres de la columna.
Ahora, discutamos cómo convertir Pyspark RDD en DataFrame.
Creación de Pyspark Rdd
En este ejemplo, crearemos un RDD llamado Estudiantes y lo mostraremos usando Action Collect ().
#Importa el módulo Pysparkimportar pyspark
#Import Sparksession para crear una sesión
de Pyspark.SQL Import Sparksession
# Importar RDD de Pyspark.RDD
de Pyspark.RDD import rdd
#cree una aplicación llamada Linuxhint
Spark_app = Sparksession.constructor.AppName ('Linuxhint').getorcreate ()
# Crear datos de estudiante con 5 filas y 6 atributos
Estudiantes = Spark_App.sparkcontext.paralelize (['rollno': '001', 'nombre': 'sravan', 'edad': 23, 'altura': 5.79, 'Peso': 67, 'Dirección': 'Guntur',
'Rollno': '002', 'Nombre': 'ojaswi', 'edad': 16, 'altura': 3.79, 'Peso': 34, 'Dirección': 'Hyd',
'Rollno': '003', 'Nombre': 'Gnanesh Chowdary', 'Age': 7, 'Altura': 2.79, 'Peso': 17, 'Dirección': 'Patna',
'rollno': '004', 'nombre': 'rohith', 'edad': 9, 'altura': 3.69, 'Peso': 28, 'Dirección': 'Hyd',
'Rollno': '005', 'Nombre': 'Sridevi', 'Age': 37, 'Altura': 5.59, 'Peso': 54, 'Dirección': 'Hyd'])
#display el rdd usando coleccionar ()
Imprimir (estudiantes.recolectar())
Producción
['Rollno': '001', 'nombre': 'sravan', 'edad': 23, 'altura': 5.79, 'Peso': 67, 'Dirección': 'Guntur','Rollno': '002', 'Nombre': 'ojaswi', 'edad': 16, 'altura': 3.79, 'Peso': 34, 'Dirección': 'Hyd',
'Rollno': '003', 'Nombre': 'Gnanesh Chowdary', 'Age': 7, 'Altura': 2.79, 'Peso': 17, 'Dirección': 'Patna',
'rollno': '004', 'nombre': 'rohith', 'edad': 9, 'altura': 3.69, 'Peso': 28, 'Dirección': 'Hyd',
'Rollno': '005', 'Nombre': 'Sridevi', 'Age': 37, 'Altura': 5.59, 'Peso': 54, 'Dirección': 'Hyd']
Método 1: Uso de CreateATATrame ()
Es posible convertir RDD en DataFrame desde una aplicación Spark con el método createAtAtArfame (). Aquí necesitamos pasar RDD a este método.
Sintaxis
Spark_app.creatataFrame (input_rdd)Donde input_rdd es el RDD.
Ejemplo
En este ejemplo, estamos convirtiendo a los estudiantes: RDD a DataFrame utilizando el método createAtAtArfame ().
importar pyspark
#Import Sparksession para crear una sesión
de Pyspark.SQL Import Sparksession
# Importar RDD de Pyspark.RDD
de Pyspark.RDD import rdd
#cree una aplicación llamada Linuxhint
Spark_app = Sparksession.constructor.AppName ('Linuxhint').getorcreate ()
# Crear datos de estudiante con 5 filas y 6 atributos
Estudiantes = Spark_App.sparkcontext.paralelize (['rollno': '001', 'nombre': 'sravan', 'edad': 23, 'altura': 5.79, 'Peso': 67, 'Dirección': 'Guntur',
'Rollno': '002', 'Nombre': 'ojaswi', 'edad': 16, 'altura': 3.79, 'Peso': 34, 'Dirección': 'Hyd',
'Rollno': '003', 'Nombre': 'Gnanesh Chowdary', 'Age': 7, 'Altura': 2.79, 'Peso': 17, 'Dirección': 'Patna',
'rollno': '004', 'nombre': 'rohith', 'edad': 9, 'altura': 3.69, 'Peso': 28, 'Dirección': 'Hyd',
'Rollno': '005', 'Nombre': 'Sridevi', 'Age': 37, 'Altura': 5.59, 'Peso': 54, 'Dirección': 'Hyd'])
#Coble el tipo de estudiantes
Imprimir (tipo (estudiantes))
#convertir RDD a DataFrame
DF = Spark_App.Createdataframe (estudiantes)
#Disprazando el marco de datos
df.espectáculo()
#Chek el tipo de DF
Imprimir (tipo (DF))
Producción
De la salida anterior, podemos ver que los estudiantes es un RDD (que se muestra usando el tipo) y después de convertir a DataFrame, mostramos el Método Show () y para la confirmación, devolvimos el tipo de DataFrame.
Método 2: Uso de creatingataFrame () con esquema
Structtype ()
Este método se utiliza para definir la estructura del marco de datos de Pyspark. Aceptará una lista de tipos de datos junto con los nombres de columnas para el marco de datos dado. Esto se conoce como el esquema del DataFrame. Almacena una colección de campos
Structfield ()
Este método se utiliza dentro del método structtype () del Pyspark DataFrame. Aceptará nombres de columnas con el tipo de datos.
Sintaxis
esquema = structtype ([[Structfield ("columna 1", tipo de datos, verdadero/falso),
Structfield ("columna 2", tipo de datos, verdadero/falso),
… ,
Structfield ("columna n", dataType, true/false)])
Donde el esquema se refiere al marco de datos cuando se crea.
Parámetros
- Structtype acepta una lista de structfields en una lista separada por una coma.
- Structfield () se usa para agregar columnas al DataFrame, que toma los nombres de columnas como el primer parámetro y el tipo de datos de las columnas particulares como el segundo parámetro.
Tenemos que usar los tipos de datos de los métodos que se importan del Pyspark.sql.Módulo de tipos.
Los tipos de datos admitidos son:
- StringType () - Se usa para almacenar valores de cadena
- Integertype () - Se utiliza para almacenar valores enteros o enteros largos
- FloatType () - Se usa para almacenar valores flotantes
- DoubleType () - Se usa para almacenar valores dobles
- Valores booleanos como tercer parámetro. Si es verdadero, se utilizará el tipo de datos dado, de lo contrario no cuando es falso.
Tenemos que pasar este esquema al método DataFrame junto con los datos.
Sintaxis
creatataFrame (datos, esquema = esquema)Sintaxis
Spark_app.creatataFrame (input_rdd)Donde, input_rdd es el RDD.
Ejemplo
En este ejemplo, estamos convirtiendo a los estudiantes - RDD en DataFrame utilizando el método createATATAFRame () con los nombres de la columna - Rollno, Nombre, edad, altura, peso y dirección
importar pyspark
#Import Sparksession para crear una sesión
de Pyspark.SQL Import Sparksession
# Importar RDD de Pyspark.RDD
de Pyspark.RDD import rdd
#e importar tipos de estructura y tipos de datos
de Pyspark.sql.tipos import structtType, structfield, stringType, integertype, floattype
#cree una aplicación llamada Linuxhint
Spark_app = Sparksession.constructor.AppName ('Linuxhint').getorcreate ()
# Crear datos de estudiante con 5 filas y 6 atributos
Estudiantes = Spark_App.sparkcontext.paralelize (['rollno': '001', 'nombre': 'sravan', 'edad': 23, 'altura': 5.79, 'Peso': 67, 'Dirección': 'Guntur',
'Rollno': '002', 'Nombre': 'ojaswi', 'edad': 16, 'altura': 3.79, 'Peso': 34, 'Dirección': 'Hyd',
'Rollno': '003', 'Nombre': 'Gnanesh Chowdary', 'Age': 7, 'Altura': 2.79, 'Peso': 17, 'Dirección': 'Patna',
'rollno': '004', 'nombre': 'rohith', 'edad': 9, 'altura': 3.69, 'Peso': 28, 'Dirección': 'Hyd',
'Rollno': '005', 'Nombre': 'Sridevi', 'Age': 37, 'Altura': 5.59, 'Peso': 54, 'Dirección': 'Hyd'])
#Coble el tipo de estudiantes
Imprimir (tipo (estudiantes))
#Define el structtype y structfields
#Para los nombres de la columna a continuación
esquema = structtype ([[
Structfield ("rollno", stringType (), verdadero),
Structfield ("name", stringType (), verdadero),
Structfield ("edad", integertype (), verdadero),
Structfield ("altura", floatType (), verdadero),
Structfield ("peso", integertype (), verdadero),
Structfield ("dirección", stringType (), verdadero)
])
#convertir RDD a DataFrame
DF = Spark_App.Createdataframe (estudiantes, esquema)
#Disprazando el marco de datos
df.espectáculo()
#Chek el tipo de DF
Imprimir (tipo (DF))
Producción
De la salida anterior, podemos ver que los estudiantes es un RDD (que se muestra usando el tipo) y después de convertir a DataFrame, mostramos el Método Show () y para la confirmación, devolvimos el tipo de DataFrame.
Método 3: Uso de TODF ()
Todf () no toma ningún parámetro y lo convierte directamente en DataFrame.
Sintaxis
input_rdd.Todf ()Donde, input_rdd es el RDD.
Ejemplo
En este ejemplo, estamos convirtiendo a los estudiantes - RDD a DataFrame usando el método TODF ().
importar pyspark
#Import Sparksession para crear una sesión
de Pyspark.SQL Import Sparksession
# Importar RDD de Pyspark.RDD
de Pyspark.RDD import rdd
#cree una aplicación llamada Linuxhint
Spark_app = Sparksession.constructor.AppName ('Linuxhint').getorcreate ()
# Crear datos de estudiante con 5 filas y 6 atributos
Estudiantes = Spark_App.sparkcontext.paralelize (['rollno': '001', 'nombre': 'sravan', 'edad': 23, 'altura': 5.79, 'Peso': 67, 'Dirección': 'Guntur',
'Rollno': '002', 'Nombre': 'ojaswi', 'edad': 16, 'altura': 3.79, 'Peso': 34, 'Dirección': 'Hyd',
'Rollno': '003', 'Nombre': 'Gnanesh Chowdary', 'Age': 7, 'Altura': 2.79, 'Peso': 17, 'Dirección': 'Patna',
'rollno': '004', 'nombre': 'rohith', 'edad': 9, 'altura': 3.69, 'Peso': 28, 'Dirección': 'Hyd',
'Rollno': '005', 'Nombre': 'Sridevi', 'Age': 37, 'Altura': 5.59, 'Peso': 54, 'Dirección': 'Hyd'])
#Coble el tipo de estudiantes
Imprimir (tipo (estudiantes))
#convertir RDD a DataFrame
DF = estudiantes.Todf ()
#Disprazando el marco de datos
df.espectáculo()
#Chek el tipo de DF
Imprimir (tipo (DF))
Producción
De la salida anterior, podemos ver que los estudiantes es un RDD (que se muestra usando el tipo) y después de convertir a DataFrame, mostramos el Método Show () y para la confirmación, devolvimos el tipo de DataFrame.
Conclusión
En este tutorial de Pyspark, vimos cómo convertir los métodos Pyspark RDD a Pyspark DataScame usando createsframe () y Todf (). Si desea proporcionar nombres de columnas explícitamente, puede usar el segundo método en este tutorial.