Github user gaborhermann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2542#discussion_r87588061
--- Diff:
flink-libraries/flink-ml/src/main/scala/org/apache/flink/ml/recommendation/ALS.scala
---
@@ -675,7 +756,69 @@ object ALS {
collector.collect((blockID, array))
}
}
- }.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0")
+ }
+
+ // broadcasting XtX matrix in the implicit case
+ val updatedFactorMatrix = if (implicitPrefs) {
+ newMatrix.withBroadcastSet(XtXtoBroadcast.get, "XtX")
+ } else {
+ newMatrix
+ }
+
+
updatedFactorMatrix.withForwardedFieldsFirst("0").withForwardedFieldsSecond("0")
+ }
+
+ /**
+ * Computes the XtX matrix for the implicit version before updating the
factors.
+ * This matrix is intended to be broadcast, but as we cannot use a sink
inside a Flink
+ * iteration, so we represent it as a [[DataSet]] with a single element
containing the matrix.
+ *
+ * The algorithm computes `X_i^T * X_i` for every block `X_i` of `X`,
+ * then sums all these computed matrices to get `X^T * X`.
+ */
+ private[recommendation] def computeXtX(x: DataSet[(Int,
Array[Array[Double]])], factors: Int):
+ DataSet[Array[Double]] = {
+ val triangleSize = factors * (factors - 1) / 2 + factors
+
+ type MtxBlock = (Int, Array[Array[Double]])
+ // construct XtX for all blocks
+ val xtx = x
+ .mapPartition(new MapPartitionFunction[MtxBlock, Array[Double]]() {
+ var xtxForBlock: Array[Double] = null
+
+ override def mapPartition(blocks: Iterable[(Int,
Array[Array[Double]])],
+ out: Collector[Array[Double]]): Unit = {
+
+ if (xtxForBlock == null) {
+ // creating the matrix if not yet created
+ xtxForBlock = Array.fill(triangleSize)(0.0)
+ } else {
+ // erasing the matrix
+ var i = 0
+ while (i < xtxForBlock.length) {
--- End diff --
Okay, you're right. I modified the code accordingly.
In this case the decision seems straightforward, but when it's not that
clear, I agree we should do profiling. Then we could see if we could make the
profiling easy to use. For now let's just keep this in mind.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
