Читать CSV в Dataframe с вложенным столбцом - PullRequest
Новая
       5

Читать CSV в Dataframe с вложенным столбцом

0 голосов
/ 11 февраля 2020

У меня есть CSV-файл, подобный этому: 

weight,animal_type,animal_interpretation
20,dog,"{is_large_animal=true, is_mammal=true}"
3.5,cat,"{is_large_animal=false, is_mammal=true}"
6.00E-04,ant,"{is_large_animal=false, is_mammal=false}"

И я создал схему класса дела со следующим: 

package types

case class AnimalsType (
                         weight: Option[Double],
                         animal_type: Option[String],
                         animal_interpretation: Option[AnimalInterpretation]
                       )

case class AnimalInterpretation (
                                  is_large_animal: Option[Boolean],
                                  is_mammal: Option[Boolean]
                                )

Я попытался загрузить CSV в кадр данных с помощью: 

var df = spark.read.format("csv").option("header", "true").load("src/main/resources/animals.csv").as[AnimalsType]

Но получил следующее исключение: 

Exception in thread "main" org.apache.spark.sql.AnalysisException: Can't extract value from animal_interpretation#12: need struct type but got string;

Я что-то не так делаю? Каков будет правильный способ сделать это?

1 Ответ

1 голос
/ 11 февраля 2020

Вы не можете назначить схему для csv json напрямую. Вам нужно преобразовать столбец csv String (animal_interpretation) в формат Json, как я сделал в приведенном ниже коде, используя UDF. если вы можете получить входные данные в формате, подобном df1, тогда вам не нужно указывать значение ниже UDF, вы можете продолжить с df1 и получить окончательный кадр данных df2.

Нет необходимости в case class, поскольку ваш заголовок данных содержит столбец, а для json данных необходимо объявить schema AnimalInterpretationSch, как показано ниже 

scala> import org.apache.spark.sql.types._

scala> import org.apache.spark.sql.expressions.UserDefinedFunction

//Input CSV DataFrame

scala> df.show(false)
+--------+-----------+---------------------------------------+
|weight  |animal_type|animal_interpretation                  |
+--------+-----------+---------------------------------------+
|20      |dog        |{is_large_animal=true, is_mammal=true} |
|3.5     |cat        |{is_large_animal=false, is_mammal=true}|
|6.00E-04|ant        |{is_large_animal=false,is_mammal=false}|
+--------+-----------+---------------------------------------+

//UDF to convert "animal_interpretation" column to Json Format

scala> def StringToJson:UserDefinedFunction = udf((data:String,JsonColumn:String) => {
     | var out = data
     | val JsonColList = JsonColumn.trim.split(",").toList
     | JsonColList.foreach{ rr => 
     | out = out.replaceAll(rr, "'"+rr+"'")
     | }
     | out = out.replaceAll("=", ":")
     | out
     | })

//All column from Json 

scala> val JsonCol = "is_large_animal,is_mammal"

//New dataframe with Json format

scala> val df1 = df.withColumn("animal_interpretation", StringToJson(col("animal_interpretation"), lit(JsonCol)))

scala> df1.show(false)
+--------+-----------+-------------------------------------------+
|weight  |animal_type|animal_interpretation                      |
+--------+-----------+-------------------------------------------+
|20      |dog        |{'is_large_animal':true, 'is_mammal':true} |
|3.5     |cat        |{'is_large_animal':false, 'is_mammal':true}|
|6.00E-04|ant        |{'is_large_animal':false,'is_mammal':false}|
+--------+-----------+-------------------------------------------+

//Schema declarion of Json format

scala> val AnimalInterpretationSch = new StructType().add("is_large_animal", BooleanType).add("is_mammal", BooleanType)

//Accessing Json columns 

scala> val df2 = df1.select(col("weight"), col("animal_type"),from_json(col("animal_interpretation"), AnimalInterpretationSch).as("jsondata")).select("weight", "animal_type", "jsondata.*")

scala> df2.printSchema
root
 |-- weight: string (nullable = true)
 |-- animal_type: string (nullable = true)
 |-- is_large_animal: boolean (nullable = true)
 |-- is_mammal: boolean (nullable = true)


scala> df2.show()
+--------+-----------+---------------+---------+
|  weight|animal_type|is_large_animal|is_mammal|
+--------+-----------+---------------+---------+
|      20|        dog|           true|     true|
|     3.5|        cat|          false|     true|
|6.00E-04|        ant|          false|    false|
+--------+-----------+---------------+---------+
...