CODEWITHSUNDEEP
scalaapache-spark
Dan
Dan

Reputation: 524

Why is Key always 0 when creating map

My code is supposed to extract a Map from a dataframe. The map will be used later for some calculations (mapping Credit to best matching original Billing). However the first step is failing already - the TransactionId is always retrieved as 0.

Simplified version of the code:

case class SalesTransaction(
                  CustomerId : Int,
                  Score : Int,
                  Revenue : Double,
                  Type : String,
                  Credited : Double = 0.0,
                  LinkedTransactionId : Int = 0,
                  IsProcessed : Boolean = false
                  )
val df = Seq(
  (1, 1, 123, "Sales", 100),
  (1, 2, 122, "Credit", 100),
  (1, 3, 99, "Sales", 70),
  (1, 4, 101, "Sales", 77),
  (1, 5, 102, "Credit", 75),
  (1, 6, 98, "Sales", 71),
  (2, 7, 200, "Sales", 55),
  (2, 8, 220, "Sales", 55),
  (2, 9, 200, "Credit", 50),
  (2, 10, 205, "Sales", 50)
).toDF("CustomerId", "TransactionId", "TransactionAttributesScore", "TransactionType", "Revenue")
    .withColumn("Revenue", $"Revenue".cast(DoubleType))
    .repartition($"CustomerId")

//map generation:
val m2 : Map[Int, SalesTransaction] =
  df.map(row => (
          row.getAs("TransactionId")
          , new SalesTransaction(row.getAs("CustomerId")
          , row.getAs("TransactionAttributesScore")
          , row.getAs("Revenue")
          , row.getAs("TransactionType")
          )
        )
  ).collect.toMap

m2.foreach(m => println("key: " + m._1 +" Value: "+ m._2))

The output has only the very last record, because all values captured by row.getAs("TransactionId") is null (i.e. translates as 0 in the m2 Map) thus tuple created in each iteration is (null, [current row SalesTransaction]).

Could you please advice me what might be wrong with my code? I'm quite new to Scala and must be missing some syntactical nuance here.

Upvotes: 0

Views: 43

Answers (2)

user238607
user238607

Reputation: 2468

You can also use row.getAs[Int]("TransactionId") as shown below :

val m2 : Map[Int, SalesTransaction] =
      df.map(row => (
        row.getAs[Int]("TransactionId"), 
        new SalesTransaction(row.getAs("CustomerId"),
                                row.getAs("TransactionAttributesScore"),
                                row.getAs("Revenue"),
                                row.getAs("TransactionType"))
                    )
      ).collect.toMap

It is always better to use the casted version of getAs to avoid errors like this.

Upvotes: 1

Dan
Dan

Reputation: 524

The issue is related to data type obtained from row.getAs("TransactionId"). Despite underlying $"TransactionId" being integer. Converting the input explicitly resolved the issue:

//… code above unchanged
val m2 : Map[Int, SlTransaction] =
  df.map(row => {
    val mKey : Int = row.getAs("TransactionId") //forcing into Int variable
    val mValue : SlTransaction = new SlTransaction(row.getAs("CustomerId")
      , row.getAs("TransactionAttributesScore")
      , row.getAs("Revenue")
      , row.getAs("TransactionType")
    )
    (mKey, mValue)
  }
  ).collect.toMap

Upvotes: 0

Related Questions