Spark 2.0 impliziter Encoder, befassen sich mit fehlender Spalte, wenn Typ ist Option [Seq [String]] (Scala) - Scala, Apache-Funke, Apache-Spark-Datensatz

Ich habe Probleme beim Codieren von Daten, wenn einige Spalten vom Typ Option [Seq [String]] in unserer Datenquelle fehlen. Idealerweise möchte ich, dass die fehlenden Spaltendaten mit gefüllt werden None.

Szenario:

Wir haben einige Parkett-Dateien, die wir lesen, die haben Spalte1 aber nicht Spalte2.

Wir laden die Daten aus diesen Parkettdateien in ein Dataset, und wirf es als MyType.

case class MyType(column1: Option[String], column2: Option[Seq[String]])

sqlContext.read.parquet("dataSource.parquet").as[MyType]

org.apache.spark.sql.AnalysisException: kann nicht aufgelöst werden "column2"angegebene Eingabespalten: [Spalte1];

Gibt es eine Möglichkeit, das Dataset mit column2-Daten als zu erstellen None?

Antworten:

In einfachen Fällen können Sie ein Anfangsschema angeben, das eine Übermenge der erwarteten Schemas darstellt. Zum Beispiel in Ihrem Fall:

val schema = Seq[MyType]().toDF.schema

Seq("a", "b", "c").map(Option(_))
.toDF("column1")
.write.parquet("/tmp/column1only")

val df = spark.read.schema(schema).parquet("/tmp/column1only").as[MyType]
df.show

+-------+-------+
|column1|column2|
+-------+-------+
|      a|   null|
|      b|   null|
|      c|   null|
+-------+-------+

df.first

MyType = MyType(Some(a),None)

Dieser Ansatz kann sein ein bisschen zerbrechlich Also sollten Sie SQL-Literale verwenden, um die Lücken zu füllen:

spark.read.parquet("/tmp/column1only")
// or ArrayType(StringType)
.withColumn("column2", lit(null).cast("array<string>"))
.as[MyType]
.first

MyType = MyType(Some(a),None)