Iz rada provjera umnoska

MultipilcationTest.scala

package hr.fer.ztel.thesis

import breeze.linalg.argtopk
import hr.fer.ztel.thesis.spark.MLlibBlockMatrixMultiplyVersion220._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.linalg.MLlibBreezeConversions._
import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, MatrixEntry}
import org.apache.spark.sql.SparkSession
import org.scalatest.{BeforeAndAfter, FlatSpec, Matchers}

class MultipilcationTest extends FlatSpec with Matchers with BeforeAndAfter {

  Logger.getLogger("org").setLevel(Level.OFF)
  Logger.getLogger("akka").setLevel(Level.OFF)

  val spark: SparkSession = SparkSession.builder.master("local[*]").getOrCreate()

  before {
    spark.newSession()
  }

  after {
    System.clearProperty("spark.driver.port")
  }

  "Multiplication" should "return good value." in {

    val userItemEntries = spark.sparkContext.makeRDD(Seq(
      (0, 0, 1), (0, 1, 2), (0, 2, 3), (0, 3, 4),
      (1, 0, 5), (1, 1, 6), (1, 2, 7), (1, 3, 8),
      (2, 0, 9), (2, 1, 1), (2, 2, 2), (2, 3, 3)
    )).map(t => MatrixEntry(t._1, t._2, t._3))

    val itemItemEntries = spark.sparkContext.makeRDD(Seq(
      (0, 0, 1), (0, 1, 2), (0, 2, 3), (0, 3, 4),
      (1, 0, 5), (1, 1, 6), (1, 2, 7), (1, 3, 8),
      (2, 0, 9), (2, 1, 1), (2, 2, 2), (2, 3, 3),
      (3, 0, 4), (3, 1, 5), (3, 2, 6), (3, 3, 7)
    )).map(t => MatrixEntry(t._1, t._2, t._3))

    val expectedRecommendations = Array(
      "0:3,0,2",
      "1:3,0,2",
      "2:3,2,0"
    )

    val numUsers = 3
    val numItems = 4

    val k = 3

    val C = new CoordinateMatrix(userItemEntries, numUsers, numItems).toBlockMatrix()
    val S = new CoordinateMatrix(itemItemEntries, numItems, numItems).toBlockMatrix()

    val R = multiply(C, S)

    val userSeenItems = userItemEntries
      .map { case MatrixEntry(user, item, _) => (user, item) }
      .groupByKey
      .map { case (user, seenItems) => (user, Set.empty[Long]) }
      .collectAsMap
      .toMap

    val userSeenItemsBroadcast = spark.sparkContext.broadcast(userSeenItems)

    val actualRecommendations = R.toIndexedRowMatrix.rows
      .mapPartitions {
        val localUserSeenItems = userSeenItemsBroadcast.value
        _.filter(row => localUserSeenItems.contains(row.index))
          .map { row =>
            val user = row.index
            val unseenItems = argtopk(row.vector.toBreeze, k)
              .filterNot(item => localUserSeenItems(user).contains(item.toLong))

            s"$user:${unseenItems.mkString(",")}"
          }
      }
      .collect

    actualRecommendations shouldEqual expectedRecommendations
  }

}