¿Cómo crear un DataFrame vacío con un esquema específico?

Quiero crear en DataFrame con un esquema especificado en Scala. Intenté usar lectura JSON (me refiero a leer el archivo vacío) pero no creo que sea la mejor práctica.

Supongamos que quiere un dataframe con el siguiente esquema:

 root |-- k: string (nullable = true) |-- v: integer (nullable = false) 

Simplemente define el esquema para un dataframe y usa un RDD[Row] vacío RDD[Row] :

 import org.apache.spark.sql.types.{ StructType, StructField, StringType, IntegerType} import org.apache.spark.sql.Row val schema = StructType( StructField("k", StringType, true) :: StructField("v", IntegerType, false) :: Nil) // Spark < 2.0 // sqlContext.createDataFrame(sc.emptyRDD[Row], schema) spark.createDataFrame(sc.emptyRDD[Row], schema) 

PySpark equivalente es casi idéntico:

 from pyspark.sql.types import StructType, StructField, IntegerType, StringType schema = StructType([ StructField("k", StringType(), True), StructField("v", IntegerType(), False) ]) # or df = sc.parallelize([]).toDF(schema) # Spark < 2.0 # sqlContext.createDataFrame([], schema) df = spark.createDataFrame([], schema) 

Usar codificadores implícitos (solo Scala) con tipos de Product como Tuple :

 import spark.implicits._ Seq.empty[(String, Int)].toDF("k", "v") 

o clase de caso:

 case class KV(k: String, v: Int) Seq.empty[KV].toDF 

o

 spark.emptyDataset[KV].toDF 

A partir de Spark 2.0.0, puede hacer lo siguiente.

Clase de caso

Definamos una clase de caso Person :

 scala> case class Person(id: Int, name: String) defined class Person 

Importar encendedores SparkSession implícitos:

 scala> import spark.implicits._ import spark.implicits._ 

Y use SparkSession para crear un Dataset[Person] vacío Dataset[Person] :

 scala> spark.emptyDataset[Person] res0: org.apache.spark.sql.Dataset[Person] = [id: int, name: string] 

Esquema DSL

También puede usar un esquema “DSL” (consulte Funciones de soporte para DataFrames en org.apache.spark.sql.ColumnName ).

 scala> val id = $"id".int id: org.apache.spark.sql.types.StructField = StructField(id,IntegerType,true) scala> val name = $"name".string name: org.apache.spark.sql.types.StructField = StructField(name,StringType,true) scala> import org.apache.spark.sql.types.StructType import org.apache.spark.sql.types.StructType scala> val mySchema = StructType(id :: name :: Nil) mySchema: org.apache.spark.sql.types.StructType = StructType(StructField(id,IntegerType,true), StructField(name,StringType,true)) scala> import org.apache.spark.sql.Row import org.apache.spark.sql.Row scala> val emptyDF = spark.createDataFrame(sc.emptyRDD[Row], mySchema) emptyDF: org.apache.spark.sql.DataFrame = [id: int, name: string] scala> emptyDF.printSchema root |-- id: integer (nullable = true) |-- name: string (nullable = true) 
 import scala.reflect.runtime.{universe => ru} def createEmptyDataFrame[T: ru.TypeTag] = hiveContext.createDataFrame(sc.emptyRDD[Row], ScalaReflection.schemaFor(ru.typeTag[T].tpe).dataType.asInstanceOf[StructType] ) case class RawData(id: String, firstname: String, lastname: String, age: Int) val sourceDF = createEmptyDataFrame[RawData] 

Aquí puede crear un esquema utilizando StructType en scala y pasar el Empty RDD para que pueda crear una tabla vacía. El siguiente código es para el mismo.

 import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.sql._ import org.apache.spark.sql.Row import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types.StructType import org.apache.spark.sql.types.StructField import org.apache.spark.sql.types.IntegerType import org.apache.spark.sql.types.BooleanType import org.apache.spark.sql.types.LongType import org.apache.spark.sql.types.StringType //import org.apache.hadoop.hive.serde2.objectinspector.StructField object EmptyTable extends App { val conf = new SparkConf; val sc = new SparkContext(conf) //create sparksession object val sparkSession = SparkSession.builder().enableHiveSupport().getOrCreate() //Created schema for three columns val schema = StructType( StructField("Emp_ID", LongType, true) :: StructField("Emp_Name", StringType, false) :: StructField("Emp_Salary", LongType, false) :: Nil) //Created Empty RDD var dataRDD = sc.emptyRDD[Row] //pass rdd and schema to create dataframe val newDFSchema = sparkSession.createDataFrame(dataRDD, schema) newDFSchema.createOrReplaceTempView("tempSchema") sparkSession.sql("create table Finaltable AS select * from tempSchema") } 

Aquí hay una solución que crea un dataframe vacío en pyspark 2.0.0 o más.

 from pyspark.sql import SQLContext sc = spark.sparkContext schema = StructType([StructField('col1', StringType(),False),StructField('col2', IntegerType(), True)]) sqlContext.createDataFrame(sc.emptyRDD(), schema) 

Versión de Java para crear DataSet vacío:

 public Dataset emptyDataSet(){ SparkSession spark = SparkSession.builder().appName("Simple Application") .config("spark.master", "local").getOrCreate(); Dataset emptyDataSet = spark.createDataFrame(new ArrayList<>(), getSchema()); return emptyDataSet; } public StructType getSchema() { String schemaString = “column1 column2 column3 column4 column5”; List fields = new ArrayList<>(); StructField indexField = DataTypes.createStructField(“column0”, DataTypes.LongType, true); fields.add(indexField); for (String fieldName : schemaString.split(" ")) { StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true); fields.add(field); } StructType schema = DataTypes.createStructType(fields); return schema; }