Wie erstelle ich einen leeren DataFrame mit einem angegebenen Schema?

94

Ich möchte DataFramemit einem bestimmten Schema in Scala erstellen . Ich habe versucht, JSON read (ich meine das Lesen leerer Dateien) zu verwenden, aber ich denke nicht, dass dies die beste Vorgehensweise ist.

user1735076
quelle

Antworten:

125

Nehmen wir an, Sie möchten einen Datenrahmen mit dem folgenden Schema:

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

Sie definieren einfach das Schema für einen Datenrahmen und verwenden leer 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)

Das PySpark-Äquivalent ist fast identisch:

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)

Verwenden impliziter Encoder (nur Scala) mit ProductTypen wie Tuple:

import spark.implicits._

Seq.empty[(String, Int)].toDF("k", "v")

oder Fallklasse:

case class KV(k: String, v: Int)

Seq.empty[KV].toDF

oder

spark.emptyDataset[KV].toDF
null323
quelle
Dies ist die am besten geeignete Antwort - vollständig und auch nützlich, wenn Sie das Schema eines vorhandenen Datasets schnell reproduzieren möchten. Ich weiß nicht, warum es nicht das akzeptierte ist.
Lucas Lima
40

Ab Spark 2.0.0 können Sie Folgendes tun.

Fallklasse

Definieren wir eine PersonFallklasse:

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

Import sparkSparkSession implizit Encoders:

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

Und verwenden Sie SparkSession, um ein leeres zu erstellen Dataset[Person]:

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

Schema DSL

Sie können auch ein Schema "DSL" verwenden (siehe Unterstützungsfunktionen für DataFrames in 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)
Jacek Laskowski
quelle
Hallo, der Compiler sagt, dass spark.emptyDatasetauf meinem Modul nicht vorhanden, wie man es benutzt? gibt es einige (richtig) ähnlich (nicht korrekt) val df = apache.spark.emptyDataset[RawData]?
Peter Krauss
@PeterKrauss sparkist der Wert, den Sie erstellen, wenn Sie SparkSession.buildernicht Teil des org.apache.sparkPakets sind. Es werden zwei sparkNamen verwendet. Es ist das, was sparkSie sofort spark-shelleinsatzbereit haben.
Jacek Laskowski
1
Danke Jacek. Ich habe korrigiert: Das SparkSession.builder-Objekt wird von der ersten allgemeinen Initialisierung als Parameter übergeben (scheint die beste Lösung zu sein) und wird jetzt ausgeführt.
Peter Krauss
3
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]
Ravindra
quelle
3

Hier können Sie ein Schema mit StructType in Scala erstellen und die leere RDD übergeben, damit Sie eine leere Tabelle erstellen können. Der folgende Code ist für das gleiche.

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")

}
Nilesh Shinde
quelle
2

Java-Version zum Erstellen eines leeren DataSet:

public Dataset<Row> emptyDataSet(){

    SparkSession spark = SparkSession.builder().appName("Simple Application")
                .config("spark.master", "local").getOrCreate();

    Dataset<Row> emptyDataSet = spark.createDataFrame(new ArrayList<>(), getSchema());

    return emptyDataSet;
}

public StructType getSchema() {

    String schemaString = "column1 column2 column3 column4 column5";

    List<StructField> 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;
}
Molay
quelle
1

Hier ist eine Lösung, die einen leeren Datenrahmen in pyspark 2.0.0 oder höher erstellt.

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

Ab Spark 2.4.3

val df = SparkSession.builder().getOrCreate().emptyDataFrame
Fuchsgeist
quelle
7
Dies löst den Schemateil der Frage nicht.
Andrew Sklyarevsky