O codificador implícito do Spark 2.0, trata da coluna ausente quando o tipo é Option [Seq [String]] (scala) - scala, apache-spark, conjunto de dados de apache-spark

Eu estou tendo alguns problemas codificando dados quando algumas colunas que são do tipo Option [Seq [String]] estão faltando em nossa fonte de dados. Idealmente, eu gostaria que os dados da coluna faltantes fossem preenchidos com None.

Cenário:

Nós temos alguns arquivos de parquete que estamos lendo em que têm coluna1 mas não coluna2.

Nós carregamos os dados desses arquivos de parquete em um Datasete lançá-lo como MyType.

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

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

org.apache.spark.sql.AnalysisException: não é possível resolver "column2"dadas colunas de entrada: [coluna1];

Existe uma maneira de criar o Dataset com dados column2 como None?

Respostas:

Em casos simples, você pode fornecer um esquema inicial que é um superconjunto de esquemas esperados. Por exemplo, no seu caso:

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)

Essa abordagem pode ser um pouco frágil então, em geral, você deve usar literais SQL para preencher os espaços em branco:

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

MyType = MyType(Some(a),None)