Github user thvasilo commented on a diff in the pull request:
https://github.com/apache/flink/pull/2542#discussion_r87421513
--- 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 --
I don't imagine this making a major difference in performance, so let's
just go with the cleaner code angle and use `fill`.
I wish we had an easy to use integrated way to do proper profiling so such
decisions can be easier (i.e. if this is 0.5% of the CPU cost, then optimizing
is pointless but right now we don't know)
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
