graphframes · SemyonSinchenko · Jan 26, 2026 · Jan 24, 2026 · Jan 26, 2026
diff --git a/core/src/main/scala/org/graphframes/lib/SVDPlusPlus.scala b/core/src/main/scala/org/graphframes/lib/SVDPlusPlus.scala
@@ -20,26 +20,77 @@ package org.graphframes.lib
 import org.apache.spark.graphframes.graphx.Edge
 import org.apache.spark.graphframes.graphx.lib as graphxlib
 import org.apache.spark.sql.DataFrame
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.functions.col
 import org.graphframes.GraphFrame
-import org.graphframes.GraphFramesUnreachableException
+import org.graphframes.InvalidGraphException
 import org.graphframes.Logging
 import org.graphframes.WithMaxIter
 
 /**
- * Implement SVD++ based on "Factorization Meets the Neighborhood: a Multifaceted Collaborative
- * Filtering Model", available at [[https://dl.acm.org/citation.cfm?id=1401944]].
+ * Arguments for SVD++ algorithm.
  *
- * Note: The status of this algorithm is EXPERIMENTAL. Its API and implementation may be changed
- * in the future.
+ * This class implements the SVD++ algorithm for Collaborative Filtering, primarily used for
+ * Recommender Systems (Link Prediction).
  *
- * The prediction rule is r,,ui,, = u + b,,u,, + b,,i,, + q,,i,,*(p,,u,, + |N(u)|^^-0.5^^*sum(y)).
- * See the details on page 6 of the article.
+ * Based on the paper "Factorization Meets the Neighborhood: a Multifaceted Collaborative
+ * Filtering Model" by Yehuda Koren (2008), available at
+ * [[https://dl.acm.org/citation.cfm?id=1401944]].
  *
- * Configuration parameters: see the description of each parameter in the article.
+ * ==Problem Definition==
+ * The algorithm predicts unknown ratings in a user-item system. It accounts for:
+ *   - Explicit preferences (user ratings).
+ *   - Implicit feedback (the history of items a user has interacted with).
+ *   - User and Item biases.
  *
- * Returns a DataFrame with vertex attributes containing the trained model. See the object
- * (static) members for the names of the output columns.
+ * The prediction rule for a rating `r_ui` (user `u`, item `i`) is:
+ * {{{
+ * r_ui = µ + b_u + b_i + q_i^T * (p_u + |N(u)|^-0.5 * sum(y_j for j in N(u)))
+ * }}}
+ * Where `N(u)` is the set of items user `u` has interacted with (implicit feedback).
+ *
+ * ==Input Requirements==
+ * !!! IMPORTANT !!! The input graph MUST be a **Directed Bipartite Graph** representing
+ * interactions:
+ *   - **Vertices**: A mix of Users and Items.
+ *   - **Edges**: Directed strictly from **User (src) -> Item (dst)**.
+ *   - **Edge Attribute**: A numeric column (default "weight") representing the rating.
+ *
+ * DO NOT use this on general/undirected graphs (e.g., social networks), as the algorithm relies
+ * on the asymmetry between Users (who provide feedback) and Items (who receive it).
+ *
+ * ==Output Model (Node Embeddings)==
+ * The algorithm returns a DataFrame of vertices with the trained model parameters. These
+ * parameters function as embeddings:
+ *
+ *   - `column1` (Array[Double]): **Primary Latent Factors (Explicit Embedding)**.
+ *   - For Users: Represents preferences (`p_u`).
+ *   - For Items: Represents characteristics (`q_i`).
+ *   - `column2` (Array[Double]): **Implicit Factors (Implicit Embedding)**.
+ *   - For Items: Represents the influence of the item (`y_i`) on a user's profile based on
+ *     viewing history.
+ *   - For Users: Generally unused/zero.
+ *   - `column3` (Double): **Bias**.
+ *   - For Users: User bias (`b_u`).
+ *   - For Items: Item bias (`b_i`).
+ *   - `column4` (Double): **Implicit Normalization Term**.
+ *   - For Users: Precomputed `|N(u)|^-0.5`.
+ *   - For Items: Unused.
+ *
+ * ==Parameter Tuning Guide==
+ *
+ * Constraints:
+ *   - `minValue` / `maxValue`: Hard bounds for predicted ratings. Predictions outside this range
+ *     are clipped. Set these to your rating scale limits (e.g., 1.0 and 5.0).
+ *
+ * Learning Rates (Step sizes for Gradient Descent):
+ *   - `gamma1`: Learning rate for **Biases** (`b_u`, `b_i`).
+ *   - `gamma2`: Learning rate for **Embeddings/Factors** (`p_u`, `q_i`, `y_j`). > Tip: Increase
+ *     if convergence is too slow. Decrease if the loss explodes (NaN).
+ *
+ * Regularization (Preventing Overfitting):
+ *   - `gamma6`: Regularization for **Biases**.
+ *   - `gamma7`: Regularization for **Embeddings/Factors**. > Tip: Increase these if the model
+ *     performs well on training data but poorly on test data.
  */
 class SVDPlusPlus private[graphframes] (private val graph: GraphFrame)
     extends Arguments
@@ -91,6 +142,11 @@ class SVDPlusPlus private[graphframes] (private val graph: GraphFrame)
   }
 
   def run(): DataFrame = {
+    import SVDPlusPlus.COLUMN_WEIGHT
+
+    if (!graph.edges.columns.contains(COLUMN_WEIGHT)) {
+      throw new InvalidGraphException(s"SVD++ requires a weight column $COLUMN_WEIGHT")
+    }
     val conf = new graphxlib.SVDPlusPlus.Conf(
       rank = _rank,
       maxIters = maxIter.getOrElse(2),
@@ -101,10 +157,37 @@ class SVDPlusPlus private[graphframes] (private val graph: GraphFrame)
       gamma6 = _gamma6,
       gamma7 = _gamma7)
 
-    val (df, l) = SVDPlusPlus.run(graph, conf)
+    val g = if (graph.hasIntegralIdType) {
+      graph
+    } else {
+      val iVertices = graph.indexedVertices
+      val iEdges = graph.indexedEdges.select(
+        col(GraphFrame.LONG_SRC).alias(GraphFrame.SRC),
+        col(GraphFrame.LONG_DST).alias(GraphFrame.DST),
+        col(GraphFrame.ATTR).getField(COLUMN_WEIGHT).alias(COLUMN_WEIGHT))
+
+      GraphFrame(iVertices, iEdges)
+    }
+
+    val (df, l) = SVDPlusPlus.run(g, conf)
+    val result = if (graph.hasIntegralIdType) {
+      df.persist()
+    } else {
+      val iV = graph.indexedVertices
+      df.withColumnRenamed(GraphFrame.ID, GraphFrame.LONG_ID)
+        .join(iV, GraphFrame.LONG_ID)
+        .drop(GraphFrame.LONG_ID)
+        .persist()
+    }
     _loss = Some(l)
+
+    // materialize
+    result.count()
+
+    // unpersist
+    df.unpersist()
     resultIsPersistent()
-    df
+    result
   }
 
   def loss: Double = {
@@ -116,9 +199,11 @@ class SVDPlusPlus private[graphframes] (private val graph: GraphFrame)
 object SVDPlusPlus {
 
   private def run(graph: GraphFrame, conf: graphxlib.SVDPlusPlus.Conf): (DataFrame, Double) = {
-    val edges = graph.edges.select(GraphFrame.SRC, GraphFrame.DST, COLUMN_WEIGHT).rdd.map {
-      case Row(src: Long, dst: Long, w: Double) => Edge(src, dst, w)
-      case _ => throw new GraphFramesUnreachableException()
+    val edges = graph.edges.select(GraphFrame.SRC, GraphFrame.DST, COLUMN_WEIGHT).rdd.map { row =>
+      val src = row.getAs[Number](0).longValue()
+      val dst = row.getAs[Number](1).longValue()
+      val w = row.getAs[Number](2).doubleValue()
+      Edge(src, dst, w)
     }
     val (gx, res) = graphxlib.SVDPlusPlus.run(edges, conf)
     val gf = GraphXConversions.fromGraphX(

diff --git a/core/src/test/scala/org/graphframes/lib/SVDPlusPlusSuite.scala b/core/src/test/scala/org/graphframes/lib/SVDPlusPlusSuite.scala
@@ -18,6 +18,7 @@
 package org.graphframes.lib
 
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.types.DataTypes
 import org.graphframes.GraphFrame
 import org.graphframes.GraphFrameTestSparkContext
@@ -55,4 +56,49 @@ class SVDPlusPlusSuite extends SparkFunSuite with GraphFrameTestSparkContext {
     assert(err <= svdppErr)
     v2.unpersist()
   }
+
+  Seq(
+    ("int", "float"),
+    ("short", "double"),
+    ("long", "float"),
+    ("byte", "double"),
+    ("string", "float")).foreach(types =>
+    test(s"Test SVD++ with mean square error on training set, ${types._1}/${types._2} types") {
+      val svdppErr = 8.0
+      val g = {
+        val gg = Graphs.ALSSyntheticData()
+        GraphFrame(
+          gg.vertices.select(col(GraphFrame.ID).cast(types._1)),
+          gg.edges.select(
+            col(GraphFrame.SRC).cast(types._1),
+            col(GraphFrame.DST).cast(types._1),
+            col("weight").cast(types._2)))
+      }
+
+      val v2 = g.svdPlusPlus.maxIter(2).run()
+      TestUtils.testSchemaInvariants(g, v2)
+      Seq(SVDPlusPlus.COLUMN1, SVDPlusPlus.COLUMN2).foreach { case c =>
+        TestUtils.checkColumnType(
+          v2.schema,
+          c,
+          DataTypes.createArrayType(DataTypes.DoubleType, false))
+      }
+      Seq(SVDPlusPlus.COLUMN3, SVDPlusPlus.COLUMN4).foreach { case c =>
+        TestUtils.checkColumnType(v2.schema, c, DataTypes.DoubleType)
+      }
+      val err = v2
+        .select(GraphFrame.ID, SVDPlusPlus.COLUMN4)
+        .rdd
+        .map { row =>
+          {
+            val vid = if (types._1 == "string") { row.getAs[String](0).toLong }
+            else { row.getAs[Number](0).longValue() }
+            val vd = row.getAs[Number](1).doubleValue()
+            if (vid % 2 == 1) vd else 0.0
+          }
+        }
+        .reduce(_ + _) / g.edges.count()
+      assert(err <= svdppErr)
+      v2.unpersist()
+    })
 }
diff --git a/python/graphframes/graphframe.py b/python/graphframes/graphframe.py
@@ -762,13 +762,51 @@ def svdPlusPlus(
         gamma6: float = 0.005,
         gamma7: float = 0.015,
     ) -> tuple[DataFrame, float]:
-        """
-        Runs the SVD++ algorithm.
-
-        See Scala documentation for more details.
-
-        :return: Tuple of DataFrame with new vertex columns storing learned model, and loss value
-        """
+        """Runs the SVD++ algorithm for Collaborative Filtering.
+
+        Based on the paper "Factorization Meets the Neighborhood: a Multifaceted Collaborative
+        Filtering Model" by Yehuda Koren (2008).
+
+        **Algorithm Description**
+        SVD++ improves upon standard Matrix Factorization by incorporating implicit feedback
+        (the history of items a user has interacted with) alongside explicit ratings.
+        The prediction rule is:
+        ``r_ui = µ + b_u + b_i + q_i^T * (p_u + |N(u)|^-0.5 * sum(y_j for j in N(u)))``
+
+        **Input Requirements**
+        The input graph must be a **Directed Bipartite Graph**:
+        - **Vertices**: A mix of Users and Items.
+        - **Edges**: Directed strictly from **User (src) -> Item (dst)**.
+        - **Edge Attribute**: Represents the rating (weight).
+
+        :param rank: The number of latent factors (embedding size).
+        :param maxIter: The maximum number of iterations.
+        :param minValue: The minimum possible rating value (used for clipping predictions).
+        :param maxValue: The maximum possible rating value (used for clipping predictions).
+        :param gamma1: Learning rate for bias parameters (`b_u`, `b_i`).
+        :param gamma2: Learning rate for factor parameters (`p_u`, `q_i`, `y_j`).
+        :param gamma6: Regularization coefficient for bias parameters.
+        :param gamma7: Regularization coefficient for factor parameters.
+        :return: A tuple ``(v, loss)`` where:
+            - ``v`` is a DataFrame of vertices containing the trained model parameters (embeddings).
+            - ``loss`` is the final training loss (double).
+
+        **Output DataFrame Columns**
+        The returned DataFrame ``v`` contains the following new columns containing the model parameters:
+
+        - **column1** (Array[Double]): Primary Latent Factors (Explicit Embedding).
+            - For Users: Preferences vector (`p_u`).
+            - For Items: Characteristics vector (`q_i`).
+        - **column2** (Array[Double]): Implicit Factors (Implicit Embedding).
+            - For Items: Influence vector (`y_i`).
+            - For Users: Unused/Zero (users aggregate `y` from neighbors).
+        - **column3** (Double): Bias term.
+            - For Users: User bias (`b_u`).
+            - For Items: Item bias (`b_i`).
+        - **column4** (Double): Implicit Normalization term.
+            - For Users: Precomputed ``|N(u)|^-0.5``.
+            - For Items: Unused.
+        """  # noqa: E501
         return self._impl.svdPlusPlus(
             rank=rank,
             maxIter=maxIter,