You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by mb...@apache.org on 2018/03/07 22:58:14 UTC
[1/2] systemml git commit: [SYSTEMML-2171] Improved caching of
ultra-sparse matrices
Repository: systemml
Updated Branches:
refs/heads/master 52b1b5716 -> 340c99d26
[SYSTEMML-2171] Improved caching of ultra-sparse matrices
This patch improves the checkpoint (i.e., distributed caching) logic for
ultra-sparse matrices. Specifically, we now account for invalid CP
dimensions not just memory to decide upon the compilation of checkpoint
instructions. Furthermore, we also avoid too eager coalesce on
checkpoints if the existing number of partitions is already below the
default parallelism of the cluster.
Project: http://git-wip-us.apache.org/repos/asf/systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/systemml/commit/5b9cb15f
Tree: http://git-wip-us.apache.org/repos/asf/systemml/tree/5b9cb15f
Diff: http://git-wip-us.apache.org/repos/asf/systemml/diff/5b9cb15f
Branch: refs/heads/master
Commit: 5b9cb15fba1e0e511d6873c1c87120f3f5af7dae
Parents: 52b1b57
Author: Matthias Boehm <mb...@gmail.com>
Authored: Wed Mar 7 00:46:49 2018 -0800
Committer: Matthias Boehm <mb...@gmail.com>
Committed: Wed Mar 7 14:58:34 2018 -0800
----------------------------------------------------------------------
src/main/java/org/apache/sysml/hops/Hop.java | 4 ++--
.../runtime/instructions/spark/CheckpointSPInstruction.java | 5 +++--
2 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/systemml/blob/5b9cb15f/src/main/java/org/apache/sysml/hops/Hop.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/Hop.java b/src/main/java/org/apache/sysml/hops/Hop.java
index 5871047..8b917b4 100644
--- a/src/main/java/org/apache/sysml/hops/Hop.java
+++ b/src/main/java/org/apache/sysml/hops/Hop.java
@@ -347,8 +347,8 @@ public abstract class Hop implements ParseInfo
//conditional checkpoint based on memory estimate in order to
//(1) avoid unnecessary persist and unpersist calls, and
//(2) avoid unnecessary creation of spark context (incl executors)
- if( OptimizerUtils.isHybridExecutionMode()
- && !OptimizerUtils.exceedsCachingThreshold(getDim2(), _outputMemEstimate)
+ if( (OptimizerUtils.isHybridExecutionMode() && hasValidCPDimsAndSize()
+ && !OptimizerUtils.exceedsCachingThreshold(getDim2(), _outputMemEstimate))
|| _etypeForced == ExecType.CP )
{
et = ExecType.CP;
http://git-wip-us.apache.org/repos/asf/systemml/blob/5b9cb15f/src/main/java/org/apache/sysml/runtime/instructions/spark/CheckpointSPInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/CheckpointSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/CheckpointSPInstruction.java
index 90f6c7b..6f8a08e 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/spark/CheckpointSPInstruction.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/CheckpointSPInstruction.java
@@ -108,8 +108,9 @@ public class CheckpointSPInstruction extends UnarySPInstruction {
//(trigger coalesce if intended number of partitions exceeded by 20%
//and not hash partitioned to avoid losing the existing partitioner)
int numPartitions = SparkUtils.getNumPreferredPartitions(mcIn, in);
- boolean coalesce = ( 1.2*numPartitions < in.getNumPartitions()
- && !SparkUtils.isHashPartitioned(in) );
+ boolean coalesce = ( 1.2*numPartitions < in.getNumPartitions()
+ && !SparkUtils.isHashPartitioned(in) && in.getNumPartitions()
+ > SparkExecutionContext.getDefaultParallelism(true));
//checkpoint pre-processing rdd operations
if( coalesce ) {
[2/2] systemml git commit: [SYSTEMML-2173] Propagation of upper bound
nnz over spark instructions
Posted by mb...@apache.org.
[SYSTEMML-2173] Propagation of upper bound nnz over spark instructions
Due to lazy evaluation many spark instructions only update the output
number of non-zeros if it can be exactly inferred from the inputs (e.g.,
for reorg operations). This is especially problematic for ultra-sparse
data sets where a treatment as dense intermediate can lead to severe
performance issues due to conservative plans and fallbacks such as
guarded collect. This patch improves the nnz handling by propagating
worst-case upper bounds at runtime level (similar to the propagation at
compiler level). Furthermore, this also includes the exploitation of
these bounds in the guarded collect path to avoid unnecessary
dense/sparse flips and HDFS I/O.
Project: http://git-wip-us.apache.org/repos/asf/systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/systemml/commit/340c99d2
Tree: http://git-wip-us.apache.org/repos/asf/systemml/tree/340c99d2
Diff: http://git-wip-us.apache.org/repos/asf/systemml/diff/340c99d2
Branch: refs/heads/master
Commit: 340c99d26d37b48b67d36402f45715b7e4bf8d85
Parents: 5b9cb15
Author: Matthias Boehm <mb...@gmail.com>
Authored: Wed Mar 7 13:13:16 2018 -0800
Committer: Matthias Boehm <mb...@gmail.com>
Committed: Wed Mar 7 14:58:36 2018 -0800
----------------------------------------------------------------------
.../org/apache/sysml/hops/OptimizerUtils.java | 2 +-
.../controlprogram/caching/MatrixObject.java | 6 ++---
.../spark/MatrixIndexingSPInstruction.java | 1 +
.../spark/MatrixReshapeSPInstruction.java | 4 +++-
.../runtime/matrix/MatrixCharacteristics.java | 24 +++++++++++++++++---
5 files changed, 29 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/systemml/blob/340c99d2/src/main/java/org/apache/sysml/hops/OptimizerUtils.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/OptimizerUtils.java b/src/main/java/org/apache/sysml/hops/OptimizerUtils.java
index b8150f8..f21fe46 100644
--- a/src/main/java/org/apache/sysml/hops/OptimizerUtils.java
+++ b/src/main/java/org/apache/sysml/hops/OptimizerUtils.java
@@ -472,7 +472,7 @@ public class OptimizerUtils
mc.getCols(),
mc.getRowsPerBlock(),
mc.getColsPerBlock(),
- mc.getNonZeros(), memPinned);
+ mc.getNonZerosBound(), memPinned);
}
public static boolean checkSparkCollectMemoryBudget( long rlen, long clen, int brlen, int bclen, long nnz, long memPinned )
http://git-wip-us.apache.org/repos/asf/systemml/blob/340c99d2/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
index e744ecb..d0d7347 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/MatrixObject.java
@@ -472,7 +472,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
int clen = (int)mc.getCols();
int brlen = (int)mc.getRowsPerBlock();
int bclen = (int)mc.getColsPerBlock();
- long nnz = mc.getNonZeros();
+ long nnz = mc.getNonZerosBound();
//guarded rdd collect
if( ii == InputInfo.BinaryBlockInputInfo && //guarded collect not for binary cell
@@ -492,11 +492,11 @@ public class MatrixObject extends CacheableData<MatrixBlock>
}
else if( ii == InputInfo.BinaryCellInputInfo ) {
//collect matrix block from binary block RDD
- mb = SparkExecutionContext.toMatrixBlock(lrdd, rlen, clen, nnz);
+ mb = SparkExecutionContext.toMatrixBlock(lrdd, rlen, clen, nnz);
}
else {
//collect matrix block from binary cell RDD
- mb = SparkExecutionContext.toMatrixBlock(lrdd, rlen, clen, brlen, bclen, nnz);
+ mb = SparkExecutionContext.toMatrixBlock(lrdd, rlen, clen, brlen, bclen, nnz);
}
}
catch(DMLRuntimeException ex) {
http://git-wip-us.apache.org/repos/asf/systemml/blob/340c99d2/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixIndexingSPInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixIndexingSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixIndexingSPInstruction.java
index d1e02ca..b6df075 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixIndexingSPInstruction.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixIndexingSPInstruction.java
@@ -99,6 +99,7 @@ public class MatrixIndexingSPInstruction extends IndexingSPInstruction {
MatrixCharacteristics mcIn = sec.getMatrixCharacteristics(input1.getName());
MatrixCharacteristics mcOut = sec.getMatrixCharacteristics(output.getName());
mcOut.set(ru-rl+1, cu-cl+1, mcIn.getRowsPerBlock(), mcIn.getColsPerBlock());
+ mcOut.setNonZerosBound(Math.min(mcOut.getLength(), mcIn.getNonZerosBound()));
checkValidOutputDimensions(mcOut);
//execute right indexing operation (partitioning-preserving if possible)
http://git-wip-us.apache.org/repos/asf/systemml/blob/340c99d2/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixReshapeSPInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixReshapeSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixReshapeSPInstruction.java
index 65b7bb7..5a80aac 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixReshapeSPInstruction.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/MatrixReshapeSPInstruction.java
@@ -92,7 +92,9 @@ public class MatrixReshapeSPInstruction extends UnarySPInstruction {
MatrixCharacteristics mcOut = sec.getMatrixCharacteristics( output.getName() );
//update output characteristics and sanity check
- mcOut.set(rows, cols, mcIn.getRowsPerBlock(), mcIn.getColsPerBlock());
+ mcOut.set(rows, cols, mcIn.getRowsPerBlock(), mcIn.getColsPerBlock(), mcIn.getNonZeros());
+ if( !mcIn.nnzKnown() )
+ mcOut.setNonZerosBound(mcIn.getNonZerosBound());
if( mcIn.getRows()*mcIn.getCols() != mcOut.getRows()*mcOut.getCols() ) {
throw new DMLRuntimeException("Incompatible matrix characteristics for reshape: "
+ mcIn.getRows()+"x"+mcIn.getCols()+" vs "+mcOut.getRows()+"x"+mcOut.getCols());
http://git-wip-us.apache.org/repos/asf/systemml/blob/340c99d2/src/main/java/org/apache/sysml/runtime/matrix/MatrixCharacteristics.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/MatrixCharacteristics.java b/src/main/java/org/apache/sysml/runtime/matrix/MatrixCharacteristics.java
index 6f73de1..7766162 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/MatrixCharacteristics.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/MatrixCharacteristics.java
@@ -76,6 +76,7 @@ public class MatrixCharacteristics implements Serializable
private int numRowsPerBlock = 1;
private int numColumnsPerBlock = 1;
private long nonZero = -1;
+ private boolean ubNnz = false;
public MatrixCharacteristics() {
@@ -106,6 +107,7 @@ public class MatrixCharacteristics implements Serializable
numRowsPerBlock = bnr;
numColumnsPerBlock = bnc;
nonZero = nnz;
+ ubNnz = false;
}
public void set(MatrixCharacteristics that) {
@@ -114,6 +116,7 @@ public class MatrixCharacteristics implements Serializable
numRowsPerBlock = that.numRowsPerBlock;
numColumnsPerBlock = that.numColumnsPerBlock;
nonZero = that.nonZero;
+ ubNnz = that.ubNnz;
}
public long getRows(){
@@ -124,6 +127,10 @@ public class MatrixCharacteristics implements Serializable
return numColumns;
}
+ public long getLength() {
+ return numRows * numColumns;
+ }
+
public int getRowsPerBlock() {
return numRowsPerBlock;
}
@@ -156,7 +163,7 @@ public class MatrixCharacteristics implements Serializable
@Override
public String toString() {
- return "["+numRows+" x "+numColumns+", nnz="+nonZero
+ return "["+numRows+" x "+numColumns+", nnz="+nonZero+" ("+ubNnz+")"
+", blocks ("+numRowsPerBlock+" x "+numColumnsPerBlock+")]";
}
@@ -175,10 +182,20 @@ public class MatrixCharacteristics implements Serializable
}
public void setNonZeros(long nnz) {
+ ubNnz = false;
nonZero = nnz;
}
public long getNonZeros() {
+ return !ubNnz ? nonZero : -1;
+ }
+
+ public void setNonZerosBound(long nnz) {
+ ubNnz = true;
+ nonZero = nnz;
+ }
+
+ public long getNonZerosBound() {
return nonZero;
}
@@ -187,7 +204,8 @@ public class MatrixCharacteristics implements Serializable
}
public boolean dimsKnown(boolean includeNnz) {
- return ( numRows >= 0 && numColumns >= 0 && (!includeNnz || nonZero >= 0));
+ return ( numRows >= 0 && numColumns >= 0
+ && (!includeNnz || nnzKnown()));
}
public boolean rowsKnown() {
@@ -199,7 +217,7 @@ public class MatrixCharacteristics implements Serializable
}
public boolean nnzKnown() {
- return ( nonZero >= 0 );
+ return ( !ubNnz && nonZero >= 0 );
}
public boolean mightHaveEmptyBlocks() {