Das Ändern der Spark-Dataframe-Spaltentypen führt zu einem Fehler

Question

Ich führe Spark in einem Jupyter-Notizbuch (mit dem Jupyter-Scala-Kernel). Ich habe einen Datenrahmen mit Spalten vom Typ String, und ich möchte einen neuen Datenrahmen mit diesen Werten als Typ Int. Ich habe alle Antworten aus diesem Beitrag versucht:

How to change column types in Spark SQL's DataFrame?.

Aber ich halte einen Fehler bekommen:

org.apache.spark.SparkException: Job aborted due to stage failure 

Insbesondere erhalte ich diese Fehlermeldung:

org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 43, Column 44: Decimal 

So ging ich und sah Linie 43:

/* 043 */ Decimal tmpDecimal6 = Decimal.apply(new java.math.BigDecimal(primitive5.toString())); 

Bisher hat nichts, was ich versucht habe, funktioniert.

Hier ist ein einfaches Beispiel:

val dF = sqlContext.load("com.databricks.spark.csv", Map("path" -> "../P80001571-ALL.csv", "header" -> "true")) 
val dF2 = castColumnTo(dF, "contbr_zip", IntegerType) 
dF2.show 


val dF = sqlContext.load("com.databricks.spark.csv", Map("path" -> 

wo castColumnTo von Martin Senne in der Post wie vorgeschlagen definiert ist oben erwähnt:

object DFHelper 
    def castColumnTo(df: DataFrame, cn: String, tpe: DataType) : DataFrame = { 
    df.withColumn(cn, df(cn).cast(tpe)) 
    } 
} 

Dies ist der Fehler:

 
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 3.0 failed 1 times, most recent failure: Lost task 0.0 in stage 3.0 (TID 3, localhost): java.util.concurrent.ExecutionException: java.lang.Exception: failed to compile: org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 97, Column 45: Decimal 

Zeile 97 sieht so aus:

Decimal tmpDecimal18 = Decimal.apply(new java.math.BigDecimal(primitive17.toString())); 

Answer 1

Ich habe das Problem offenbar gelöst; Es hing damit zusammen, wie ich Spark eingerichtet hatte, um es im Notebook laufen zu lassen.

Das ist, was ich hatte vor:

@transient val Spark = new ammonite.spark.Spark 

import Spark.{ sparkConf, sc, sqlContext } 
sc 
import sqlContext.implicits._ 
import sqlContext._ 

Das ist, was ich jetzt haben:

jetzt

import org.apache.spark.SparkConf 
import org.apache.spark.SparkContext 
val conf = new SparkConf().setAppName("appname").setMaster("local") 
val sc = new SparkContext(conf) 
import org.apache.spark.sql.SQLContext 
val sqlContext = new SQLContext(sc) 

sqlContext 
import sqlContext._ 
import sqlContext.implicits._ 

Dinge scheinen zu arbeiten.