Pfthb

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.

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.

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