Para pasar el esquema a un archivo json hacemos esto:
from pyspark.sql.types import (StructField, StringType, StructType, IntegerType) data_schema = [StructField('age', IntegerType(), True), StructField('name', StringType(), True)] final_struc = StructType(fields = data_schema) df =spark.read.json('people.json', schema=final_struc)El código anterior funciona como se esperaba. Sin embargo, ahora tengo datos en la tabla que muestro por:
df = sqlContext.sql("SELECT * FROM people_json")Pero si trato de pasarle un nuevo esquema usando el siguiente comando, no funciona.
df2 = spark.sql("SELECT * FROM people_json", schema=final_struc)Da el siguiente error:
sql() obtuvo un argumento de palabra clave inesperado 'esquema'
NOTA: estoy usando Databrics Community Edition
No puede aplicar un nuevo esquema a un marco de datos ya creado. Sin embargo, puede cambiar el esquema de cada columna convirtiendo a otro tipo de datos como se muestra a continuación.
df.withColumn("column_name", $"column_name".cast("new_datatype"))Si necesita aplicar un nuevo esquema, debe convertir a RDD y crear un nuevo marco de datos nuevamente como se muestra a continuación
df = sqlContext.sql("SELECT * FROM people_json") val newDF = spark.createDataFrame(df.rdd, schema=schema)¡Espero que esto ayude!
Ya hay una respuesta disponible, pero aún quiero agregar algo.
usando toDF
newDf = rdd.toDF(schema, column_name_list)
usando createDataFrame
newDF = spark.createDataFrame(rdd ,schema, [list_of_column_name])
supongamos que tengo DataFrame con columnas|tipo de datos - nombre|cadena, marcas|cadena, género|cadena.
si quiero obtener solo marcas como enteros.
newDF = oldDF.select("marks") newDF_with_int = newDF.withColumn("marks", df['marks'].cast('Integer'))Esto convertirá las marcas en enteros.