Spark - Schema programmgesteuert mit verschiedenen Datentypen erstellen

Question

Ich habe ein Dataset bestehend aus 7-8 Felder des Typs String, Int & Float.

Am versuchen Schema von programmatischen Ansatz zu schaffen, indem diese mit:

val schema = StructType(header.split(",").map(column => StructField(column, StringType, true))) 

und Kartierung es dann Typ Zeile wie:

val dataRdd = datafile.filter(x => x!=header).map(x => x.split(",")).map(col => Row(col(0).trim, col(1).toInt, col(2).toFloat, col(3), col(4) ,col(5), col(6), col(7), col(8))) 

Aber nach Datenrahmen zu schaffen, wenn ich verwende DF.show () Es gibt Fehler für das Integer-Feld.

So wie solches Schema zu schaffen, in dem wir mehrere Datentypen im Datensatz

Answer 1

Das Problem, das Sie in Ihrem Code haben müssen, ist, dass Sie all Felder wie String zuweisen.

Vorausgesetzt, dass Sie in der Kopfzeile nur den Namen der Felder haben, können Sie den Typ nicht erraten.

Nehmen wir an, dass die Header-Zeichenfolge wie folgt

ist

val header = "field1:Int,field2:Double,field3:String" 

Dann sollte der Code

def inferType(field: String) = field.split(":")(1) match { 
    case "Int" => IntegerType 
    case "Double" => DoubleType 
    case "String" => StringType 
    case _ => StringType 
} 

val schema = StructType(header.split(",").map(column => StructField(column, inferType(column), true))) 

Für die Header-String Beispiel Sie

root 
|-- field1:Int: integer (nullable = true) 
|-- field2:Double: double (nullable = true) 
|-- field3:String: string (nullable = true) 

Auf der anderen Seite erhalten werden . Wenn es sich um einen Datenrahmen aus Text handelt, würde ich vorschlagen, dass Sie den Datenrahmen direkt aus der Datei selbst erstellen. Es ist sinnlos, es von einer RDD zu erstellen.

val fileReader = spark.read.format("com.databricks.spark.csv") 
    .option("mode", "DROPMALFORMED") 
    .option("header", "true") 
    .option("inferschema", "true") 
    .option("delimiter", ",") 

val df = fileReader.load(PATH_TO_FILE) 

Answer 2

der Strukturtyp Definieren erster:

val schema1 = StructType(Array(
    StructField("AcutionId", StringType, true), 
    StructField("Bid", IntegerType, false), 
    StructField("BidTime", FloatType, false), 
    StructField("Bidder", StringType, true), 
    StructField("BidderRate", FloatType, false), 
    StructField("OpenBid", FloatType, false), 
    StructField("Price", FloatType, false), 
    StructField("Item", StringType, true), 
    StructField("DaystoLive", IntegerType, false) 
)) 

dann jede Spalte angeben, die es sich in einer Reihe vorhanden b wird auf bestimmte Typen durch Umwandlung:

val dataRdd = datafile.filter(x => x!=header).map(x => x.split(",")) 
    .map(col => Row(
    col(0).trim, 
    col(1).trim.toInt, 
    col(2).trim.toFloat, 
    col(3).trim, 
    col(4).trim.toFloat, 
    col(5).trim.toFloat, 
    col(6).trim.toFloat, 
    col(7).trim, 
    col(8).trim.toInt) 
) 

Dann wird die Anwendung Schema zum RDD

val auctionDF = spark.sqlContext.createDataFrame(dataRdd,schema1)