Spark Scala создает dataFrame из rdd, используя Row и Schema - PullRequest
Новая
       6

Spark Scala создает dataFrame из rdd, используя Row и Schema

0 голосов
/ 12 сентября 2018

Я пытаюсь создать фрейм данных из RDD, чтобы иметь возможность записи в json со следующим форматом Пример json, как показано ниже (ожидаемый результат)

"1234": [ { LOC: 'ABC', cost1: 1,234, cost2: 2,3445 }, { LOC: 'WWW', cost1: 1,534, cost2: 6,3445 } ]

Я могу сгенерировать JSON с cost1 и cost2 в формате String. Но я хочу, чтобы cost1 и cost2 были удвоены. Я получаю сообщение об ошибке при создании фрейма данных из rdd с использованием определенной схемы. Каким-то образом данные рассматриваются как String вместо double. Может ли кто-нибудь помочь мне понять это правильно? Ниже приведен мой scala-код моего примера реализации 

object csv2json {
  def f[T](v: T) = v match {
  case _: Int    => "Int"
  case _: String => "String"
  case _: Float => "Float"
  case _: Double => "Double"
  case _:BigDecimal => "BigDecimal"
  case _         => "Unknown"
  }

  def main(args: Array[String]): Unit = {
    import org.apache.spark.sql.SparkSession
    val spark = SparkSession.builder().master("local").getOrCreate()
    import spark.implicits._

    val input_df = Seq(("12345", "111","1.34","2.34"),("123456", "112","1.343","2.344"),("1234", "113","1.353","2.354"),("1231", "114","5.343","6.344")).toDF("item_id","loc","cost1","cost2")
    input_df.show()  
    val inputRDD =  input_df.rdd.map(data => {

        val  nodeObj  = scala.collection.immutable.Map("nodeId" -> data(1).toString()
        ,"soc" -> data(2).toString().toDouble
        ,"mdc" -> data(3).toString().toDouble)
        (data(0).toString(),nodeObj)
      })

      val inputRDDAgg = inputRDD.aggregateByKey(scala.collection.mutable.ListBuffer.empty[Any])((nodeAAggreg,costValue) => nodeAAggreg += costValue , (nodeAAggreg,costValue) => nodeAAggreg ++ costValue)

      val inputRDDAggRow = inputRDDAgg.map(data => {
        println(data._1 + "and------ " + f(data._1))
        println(data._2 + "and------ " + f(data._2))

        val  skuObj  = Row(
         data._1,
         data._2)
        skuObj
      }
      )

      val innerSchema =  ArrayType(MapType(StringType, DoubleType, true))
      val schema:StructType = StructType(Seq(StructField(name="skuId", dataType=StringType),StructField(name="nodes", innerSchema)))
      val finalJsonDF = spark.createDataFrame(inputRDDAggRow, schema)
      finalJsonDF.show()
  }
}

Ниже приведена трассировка стека исключений: 

java.lang.RuntimeException: Error while encoding: java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Double
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 0, skuId), StringType), true, false) AS skuId#32
if (assertnotnull(input[0, org.apache.spark.sql.Row, true]).isNullAt) null else mapobjects(MapObjects_loopValue0, MapObjects_loopIsNull0, ObjectType(class java.lang.Object), if (isnull(validateexternaltype(lambdavariable(MapObjects_loopValue0, MapObjects_loopIsNull0, ObjectType(class java.lang.Object), true), MapType(StringType,DoubleType,true)))) null else newInstance(class org.apache.spark.sql.catalyst.util.ArrayBasedMapData), validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true]), 1, nodes), ArrayType(MapType(StringType,DoubleType,true),true)), None) AS nodes#33
    at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:291)
    at org.apache.spark.sql.SparkSession$$anonfun$4.apply(SparkSession.scala:589)
    at org.apache.spark.sql.SparkSession$$anonfun$4.apply(SparkSession.scala:589)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)

Ответы [ 2 ]

0 голосов
/ 12 сентября 2018

Я вижу несоответствие схемы в вашем коде. Я сделал простое исправление в качестве обходного пути

Я конвертировал data(1).toString в data(1).toString.toDouble. В вашем ArrayType(MapType(StringType, DoubleType, true)) вы упомянули все значения Double, где одно из ваших значений - String. Я считаю, что это проблема. 

val inputRDD =  input_df.rdd.map(data => {

      val  nodeObj  = scala.collection.immutable.Map("nodeId" -> data(1).toString.toDouble
        ,"soc" -> data(2).toString().toDouble
        ,"mdc" -> data(3).toString().toDouble)
      (data(0).toString(),nodeObj)
    })

Выход 

+------+--------------------------------------------------+
|skuId |nodes                                             |
+------+--------------------------------------------------+
|1231  |[Map(nodeId -> 114.0, soc -> 5.343, mdc -> 6.344)]|
|12345 |[Map(nodeId -> 111.0, soc -> 1.34, mdc -> 2.34)]  |
|123456|[Map(nodeId -> 112.0, soc -> 1.343, mdc -> 2.344)]|
|1234  |[Map(nodeId -> 113.0, soc -> 1.353, mdc -> 2.354)]|
+------+--------------------------------------------------+

Надеюсь, это поможет!

0 голосов
/ 12 сентября 2018

Я бы посоветовал вам использовать набор данных или фрейм данных, используя функции inbult, поскольку они являются оптимизированной версией rdds.

Таким образом, вы можете сделать следующее, чтобы удовлетворить ваши требования 

import org.apache.spark.sql.functions._
val finalJsonDF = input_df
  .groupBy("item_id")
  .agg(
    collect_list(
      struct(col("loc"), col("cost1").cast("double"), col("cost2").cast("double")))
      .as("jsonData"))

, где collect_list и struct - встроенные функции

, который должен дать вам 

+-------+-------------------+
|item_id|jsonData           |
+-------+-------------------+
|123456 |[[112,1.343,2.344]]|
|1234   |[[113,1.353,2.354]]|
|1231   |[[114,5.343,6.344]]|
|12345  |[[111,1.34,2.34]]  |
+-------+-------------------+

и сохранение jsonData в файл json как ваше требование 

finalJsonDF.coalesce(1).write.json("path to output file")

должен дать вам 

{"item_id":"123456","jsonData":[{"loc":"112","col2":1.343,"col3":2.344}]}
{"item_id":"1234","jsonData":[{"loc":"113","col2":1.353,"col3":2.354}]}
{"item_id":"1231","jsonData":[{"loc":"114","col2":5.343,"col3":6.344}]}
{"item_id":"12345","jsonData":[{"loc":"111","col2":1.34,"col3":2.34}]}
...