[SYSTEMML-567] Cleanup execution context matrix/frame object handling Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/2a0e4102 Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/2a0e4102 Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/2a0e4102
Branch: refs/heads/master Commit: 2a0e4102926623b6151d886afb40ae816b2e13d3 Parents: 7fb1117 Author: Matthias Boehm <[email protected]> Authored: Sun Jun 5 00:08:00 2016 -0700 Committer: Matthias Boehm <[email protected]> Committed: Sun Jun 5 00:08:00 2016 -0700 ---------------------------------------------------------------------- .../controlprogram/ParForProgramBlock.java | 8 ++++---- .../controlprogram/context/ExecutionContext.java | 18 +++++++++--------- .../parfor/RemoteDPParForSparkWorker.java | 2 +- .../parfor/RemoteDPParWorkerReducer.java | 8 ++++---- .../cp/DataPartitionCPInstruction.java | 2 +- .../cp/MatrixBuiltinCPInstruction.java | 3 +-- .../cp/MatrixIndexingCPInstruction.java | 2 +- .../cp/MatrixMatrixBuiltinCPInstruction.java | 2 +- .../cp/MultiReturnBuiltinCPInstruction.java | 2 +- .../cp/ParameterizedBuiltinCPInstruction.java | 4 ++-- .../cp/QuantilePickCPInstruction.java | 2 +- .../instructions/cp/VariableCPInstruction.java | 8 ++++---- .../cpfile/MatrixIndexingCPFileInstruction.java | 4 ++-- .../ParameterizedBuiltinCPFileInstruction.java | 4 ++-- .../spark/ParameterizedBuiltinSPInstruction.java | 7 +++---- .../instructions/spark/WriteSPInstruction.java | 3 +-- 16 files changed, 38 insertions(+), 41 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/controlprogram/ParForProgramBlock.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/ParForProgramBlock.java b/src/main/java/org/apache/sysml/runtime/controlprogram/ParForProgramBlock.java index d03d3b8..e9aef4b 100644 --- a/src/main/java/org/apache/sysml/runtime/controlprogram/ParForProgramBlock.java +++ b/src/main/java/org/apache/sysml/runtime/controlprogram/ParForProgramBlock.java @@ -880,7 +880,7 @@ public class ParForProgramBlock extends ForProgramBlock exportMatricesToHDFS(ec); // Step 3) submit MR job (wait for finished work) - MatrixObject colocatedDPMatrixObj = (_colocatedDPMatrix!=null)? (MatrixObject)ec.getVariable(_colocatedDPMatrix) : null; + MatrixObject colocatedDPMatrixObj = (_colocatedDPMatrix!=null)? ec.getMatrixObject(_colocatedDPMatrix) : null; RemoteParForJobReturn ret = RemoteParForMR.runJob(_ID, program, taskFile, resultFile, colocatedDPMatrixObj, _enableCPCaching, _numThreads, WRITE_REPLICATION_FACTOR, MAX_RETRYS_ON_ERROR, getMinMemory(ec), (ALLOW_REUSE_MR_JVMS & _jvmReuse) ); @@ -935,7 +935,7 @@ public class ParForProgramBlock extends ForProgramBlock // Step 1) prepare partitioned input matrix (needs to happen before serializing the progam) ParForStatementBlock sb = (ParForStatementBlock) getStatementBlock(); - MatrixObject inputMatrix = (MatrixObject)ec.getVariable(_colocatedDPMatrix ); + MatrixObject inputMatrix = ec.getMatrixObject(_colocatedDPMatrix ); PDataPartitionFormat inputDPF = sb.determineDataPartitionFormat( _colocatedDPMatrix ); inputMatrix.setPartitioned(inputDPF, 1); //mark matrix var as partitioned (for reducers) @@ -1071,7 +1071,7 @@ public class ParForProgramBlock extends ForProgramBlock // Step 1) prepare partitioned input matrix (needs to happen before serializing the progam) ParForStatementBlock sb = (ParForStatementBlock) getStatementBlock(); - MatrixObject inputMatrix = (MatrixObject)ec.getVariable(_colocatedDPMatrix ); + MatrixObject inputMatrix = ec.getMatrixObject(_colocatedDPMatrix ); PDataPartitionFormat inputDPF = sb.determineDataPartitionFormat( _colocatedDPMatrix ); inputMatrix.setPartitioned(inputDPF, 1); //mark matrix var as partitioned (for reducers) @@ -2093,7 +2093,7 @@ public class ParForProgramBlock extends ForProgramBlock MatrixObject out = null; synchronized( _ec.getVariables() ){ - out = (MatrixObject) _ec.getVariable(varname); + out = _ec.getMatrixObject(varname); } MatrixObject[] in = new MatrixObject[ _refVars.length ]; http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/controlprogram/context/ExecutionContext.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/context/ExecutionContext.java b/src/main/java/org/apache/sysml/runtime/controlprogram/context/ExecutionContext.java index 3536cd8..cc4fcf2 100644 --- a/src/main/java/org/apache/sysml/runtime/controlprogram/context/ExecutionContext.java +++ b/src/main/java/org/apache/sysml/runtime/controlprogram/context/ExecutionContext.java @@ -210,7 +210,7 @@ public class ExecutionContext public MatrixBlock getMatrixInput(String varName) throws DMLRuntimeException { - MatrixObject mo = (MatrixObject) getVariable(varName); + MatrixObject mo = getMatrixObject(varName); return mo.acquireRead(); } @@ -223,7 +223,7 @@ public class ExecutionContext public void releaseMatrixInput(String varName) throws DMLRuntimeException { - MatrixObject mo = (MatrixObject) getVariable(varName); + MatrixObject mo = getMatrixObject(varName); mo.release(); } @@ -237,7 +237,7 @@ public class ExecutionContext public FrameBlock getFrameInput(String varName) throws DMLRuntimeException { - FrameObject fo = (FrameObject) getVariable(varName); + FrameObject fo = getFrameObject(varName); return fo.acquireRead(); } @@ -250,7 +250,7 @@ public class ExecutionContext public void releaseFrameInput(String varName) throws DMLRuntimeException { - FrameObject fo = (FrameObject) getVariable(varName); + FrameObject fo = getFrameObject(varName); fo.release(); } @@ -302,8 +302,8 @@ public class ExecutionContext public void setMatrixOutput(String varName, MatrixBlock outputData) throws DMLRuntimeException { - MatrixObject mo = (MatrixObject) getVariable(varName); - mo.acquireModify(outputData); + MatrixObject mo = getMatrixObject(varName); + mo.acquireModify(outputData); mo.release(); setVariable(varName, mo); @@ -321,8 +321,8 @@ public class ExecutionContext { if( flag.isInPlace() ) { //modify metadata to carry update status - MatrixObject sores = (MatrixObject) this.getVariable (varName); - sores.setUpdateType( flag ); + MatrixObject mo = getMatrixObject(varName); + mo.setUpdateType( flag ); } //default case @@ -338,7 +338,7 @@ public class ExecutionContext public void setFrameOutput(String varName, FrameBlock outputData) throws DMLRuntimeException { - FrameObject fo = (FrameObject) getVariable(varName); + FrameObject fo = getFrameObject(varName); if( outputData.getNumColumns()>0 && outputData.getSchema()!=null ) fo.setValueType(outputData.getSchema().get(0)); fo.acquireModify(outputData); http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParForSparkWorker.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParForSparkWorker.java b/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParForSparkWorker.java index 7ee578f..318a71e 100644 --- a/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParForSparkWorker.java +++ b/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParForSparkWorker.java @@ -118,7 +118,7 @@ public class RemoteDPParForSparkWorker extends ParWorker implements PairFlatMapF partition = collectBinaryCellInput( larg._2() ); //update in-memory matrix partition - MatrixObject mo = (MatrixObject)_ec.getVariable( _inputVar ); + MatrixObject mo = _ec.getMatrixObject( _inputVar ); mo.setInMemoryPartition( partition ); //create tasks for input data http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParWorkerReducer.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParWorkerReducer.java b/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParWorkerReducer.java index 2bcc1ad..3596c1e 100644 --- a/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParWorkerReducer.java +++ b/src/main/java/org/apache/sysml/runtime/controlprogram/parfor/RemoteDPParWorkerReducer.java @@ -100,14 +100,14 @@ public class RemoteDPParWorkerReducer extends ParWorker _partition = collectBinaryBlock( valueList ); else _partition = collectBinaryCellInput( valueList ); - - //update in-memory matrix partition - MatrixObject mo = (MatrixObject)_ec.getVariable( _inputVar ); - mo.setInMemoryPartition( _partition ); //execute program LOG.trace("execute RemoteDPParWorkerReducer "+_stringID+" ("+_workerID+")"); try { + //update in-memory matrix partition + MatrixObject mo = _ec.getMatrixObject( _inputVar ); + mo.setInMemoryPartition( _partition ); + //create tasks for input data Task lTask = new Task(TaskType.SET); lTask.addIteration( new IntObject(_iterVar,key.get()) ); http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/DataPartitionCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/DataPartitionCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/DataPartitionCPInstruction.java index e474bcd..58ebcb1 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/DataPartitionCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/DataPartitionCPInstruction.java @@ -81,7 +81,7 @@ public class DataPartitionCPInstruction extends UnaryCPInstruction throws DMLRuntimeException { //get input - MatrixObject moIn = (MatrixObject) ec.getVariable(input1.getName()); + MatrixObject moIn = ec.getMatrixObject(input1.getName()); MatrixBlock mb = moIn.acquireRead(); //execute operations http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixBuiltinCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixBuiltinCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixBuiltinCPInstruction.java index 6116bc1..003b85d 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixBuiltinCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixBuiltinCPInstruction.java @@ -20,7 +20,6 @@ package org.apache.sysml.runtime.instructions.cp; import org.apache.sysml.runtime.DMLRuntimeException; -import org.apache.sysml.runtime.controlprogram.caching.MatrixObject; import org.apache.sysml.runtime.controlprogram.context.ExecutionContext; import org.apache.sysml.runtime.matrix.data.LibCommonsMath; import org.apache.sysml.runtime.matrix.data.MatrixBlock; @@ -43,7 +42,7 @@ public class MatrixBuiltinCPInstruction extends BuiltinUnaryCPInstruction String opcode = getOpcode(); if(LibCommonsMath.isSupportedUnaryOperation(opcode)) { - MatrixBlock retBlock = LibCommonsMath.unaryOperations((MatrixObject)ec.getVariable(input1.getName()),getOpcode()); + MatrixBlock retBlock = LibCommonsMath.unaryOperations(ec.getMatrixObject(input1.getName()),getOpcode()); ec.setMatrixOutput(output_name, retBlock); } else { http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixIndexingCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixIndexingCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixIndexingCPInstruction.java index 0d21626..4d99a32 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixIndexingCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixIndexingCPInstruction.java @@ -49,7 +49,7 @@ public final class MatrixIndexingCPInstruction extends IndexingCPInstruction IndexRange ixrange = getIndexRange(ec); //get original matrix - MatrixObject mo = (MatrixObject)ec.getVariable(input1.getName()); + MatrixObject mo = ec.getMatrixObject(input1.getName()); //right indexing if( opcode.equalsIgnoreCase("rangeReIndex") ) http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixMatrixBuiltinCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixMatrixBuiltinCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixMatrixBuiltinCPInstruction.java index 1fd0eb9..daba3bc 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixMatrixBuiltinCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/MatrixMatrixBuiltinCPInstruction.java @@ -47,7 +47,7 @@ public class MatrixMatrixBuiltinCPInstruction extends BuiltinBinaryCPInstruction String opcode = getOpcode(); if ( LibCommonsMath.isSupportedMatrixMatrixOperation(opcode) ) { - MatrixBlock solution = LibCommonsMath.matrixMatrixOperations((MatrixObject)ec.getVariable(input1.getName()), (MatrixObject)ec.getVariable(input2.getName()), opcode); + MatrixBlock solution = LibCommonsMath.matrixMatrixOperations(ec.getMatrixObject(input1.getName()), (MatrixObject)ec.getVariable(input2.getName()), opcode); ec.setMatrixOutput(output.getName(), solution); return; } http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/MultiReturnBuiltinCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/MultiReturnBuiltinCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/MultiReturnBuiltinCPInstruction.java index 5fc61cc..17f9350 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/MultiReturnBuiltinCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/MultiReturnBuiltinCPInstruction.java @@ -101,7 +101,7 @@ public class MultiReturnBuiltinCPInstruction extends ComputationCPInstruction throws DMLRuntimeException { String opcode = getOpcode(); - MatrixObject mo = (MatrixObject) ec.getVariable(input1.getName()); + MatrixObject mo = ec.getMatrixObject(input1.getName()); MatrixBlock[] out = null; if(LibCommonsMath.isSupportedMultiReturnOperation(opcode)) http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/ParameterizedBuiltinCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/ParameterizedBuiltinCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/ParameterizedBuiltinCPInstruction.java index 12bfbc6..70e7a11 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/ParameterizedBuiltinCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/ParameterizedBuiltinCPInstruction.java @@ -239,8 +239,8 @@ public class ParameterizedBuiltinCPInstruction extends ComputationCPInstruction ec.releaseMatrixInput(params.get("target")); } else if ( opcode.equalsIgnoreCase("transform")) { - FrameObject fo = (FrameObject) ec.getVariable(params.get("target")); - MatrixObject out = (MatrixObject) ec.getVariable(output.getName()); + FrameObject fo = ec.getFrameObject(params.get("target")); + MatrixObject out = ec.getMatrixObject(output.getName()); try { JobReturn jt = DataTransform.cpDataTransform(this, new FrameObject[] { fo } , new MatrixObject[] {out} ); out.updateMatrixCharacteristics(jt.getMatrixCharacteristics(0)); http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/QuantilePickCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/QuantilePickCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/QuantilePickCPInstruction.java index f061b22..b1f827d 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/QuantilePickCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/QuantilePickCPInstruction.java @@ -129,7 +129,7 @@ public class QuantilePickCPInstruction extends BinaryCPInstruction } else //MR VALUEPICK { - MatrixObject mat = (MatrixObject)ec.getVariable(input1.getName()); + MatrixObject mat = ec.getMatrixObject(input1.getName()); String fname = mat.getFileName(); MetaData mdata = mat.getMetaData(); ScalarObject pickindex = ec.getScalarInput(input2.getName(), input2.getValueType(), input2.isLiteral()); http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cp/VariableCPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cp/VariableCPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cp/VariableCPInstruction.java index 0fc2706..5d01c85 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cp/VariableCPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cp/VariableCPInstruction.java @@ -736,13 +736,13 @@ public class VariableCPInstruction extends CPInstruction writeCSVFile(ec, fname); else { // Default behavior - MatrixObject mo = (MatrixObject)ec.getVariable(input1.getName()); + MatrixObject mo = ec.getMatrixObject(input1.getName()); mo.exportData(fname, outFmt); } } else if( input1.getDataType() == DataType.FRAME ) { String outFmt = input3.getName(); - FrameObject mo = (FrameObject)ec.getVariable(input1.getName()); + FrameObject mo = ec.getFrameObject(input1.getName()); mo.exportData(fname, outFmt); } } @@ -779,7 +779,7 @@ public class VariableCPInstruction extends CPInstruction private void writeCSVFile(ExecutionContext ec, String fname) throws DMLRuntimeException { - MatrixObject mo = (MatrixObject)ec.getVariable(input1.getName()); + MatrixObject mo = ec.getMatrixObject(input1.getName()); String outFmt = "csv"; if(mo.isDirty()) { @@ -818,7 +818,7 @@ public class VariableCPInstruction extends CPInstruction private void writeMMFile(ExecutionContext ec, String fname) throws DMLRuntimeException { - MatrixObject mo = (MatrixObject)ec.getVariable(input1.getName()); + MatrixObject mo = ec.getMatrixObject(input1.getName()); String outFmt = "matrixmarket"; if(mo.isDirty()) { // there exist data computed in CP that is not backed up on HDFS http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cpfile/MatrixIndexingCPFileInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cpfile/MatrixIndexingCPFileInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cpfile/MatrixIndexingCPFileInstruction.java index d078560..4481e28 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cpfile/MatrixIndexingCPFileInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cpfile/MatrixIndexingCPFileInstruction.java @@ -96,7 +96,7 @@ public final class MatrixIndexingCPFileInstruction extends IndexingCPInstruction { String opcode = getOpcode(); IndexRange ixrange = getIndexRange(ec).add(1); - MatrixObject mo = (MatrixObject) ec.getVariable(input1.getName()); + MatrixObject mo = ec.getMatrixObject(input1.getName()); if( mo.isPartitioned() && opcode.equalsIgnoreCase("rangeReIndex") ) { @@ -106,7 +106,7 @@ public final class MatrixIndexingCPFileInstruction extends IndexingCPInstruction if( MapReduceTool.existsFileOnHDFS(pfname) ) { - MatrixObject out = (MatrixObject)ec.getVariable(output.getName()); + MatrixObject out = ec.getMatrixObject(output.getName()); //create output matrix object MatrixObject mobj = new MatrixObject(mo.getValueType(), pfname ); http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/cpfile/ParameterizedBuiltinCPFileInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/cpfile/ParameterizedBuiltinCPFileInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/cpfile/ParameterizedBuiltinCPFileInstruction.java index 5302736..7fb1b88 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/cpfile/ParameterizedBuiltinCPFileInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/cpfile/ParameterizedBuiltinCPFileInstruction.java @@ -122,8 +122,8 @@ public class ParameterizedBuiltinCPFileInstruction extends ParameterizedBuiltinC if ( opcode.equalsIgnoreCase("rmempty") ) { // get inputs - MatrixObject src = (MatrixObject)ec.getVariable( params.get("target") ); - MatrixObject out = (MatrixObject)ec.getVariable( output.getName() ); + MatrixObject src = ec.getMatrixObject( params.get("target") ); + MatrixObject out = ec.getMatrixObject( output.getName() ); String margin = params.get("margin"); // export input matrix (if necessary) http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java index 986d8e8..c09c1fc 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java @@ -394,10 +394,9 @@ public class ParameterizedBuiltinSPInstruction extends ComputationSPInstruction { // perform data transform on Spark try { - DataTransform.spDataTransform( - this, - new FrameObject[] { (FrameObject) sec.getVariable(params.get("target")) }, - new MatrixObject[] { (MatrixObject) sec.getVariable(output.getName()) }, ec); + DataTransform.spDataTransform( this, + new FrameObject[] { sec.getFrameObject(params.get("target")) }, + new MatrixObject[] { sec.getMatrixObject(output.getName()) }, ec); } catch (Exception e) { throw new DMLRuntimeException(e); } http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2a0e4102/src/main/java/org/apache/sysml/runtime/instructions/spark/WriteSPInstruction.java ---------------------------------------------------------------------- diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/WriteSPInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/WriteSPInstruction.java index 987cf7d..3ea03ef 100644 --- a/src/main/java/org/apache/sysml/runtime/instructions/spark/WriteSPInstruction.java +++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/WriteSPInstruction.java @@ -30,7 +30,6 @@ import org.apache.spark.api.java.JavaRDD; import org.apache.sysml.parser.Expression.ValueType; import org.apache.sysml.runtime.DMLRuntimeException; -import org.apache.sysml.runtime.controlprogram.caching.MatrixObject; import org.apache.sysml.runtime.controlprogram.context.ExecutionContext; import org.apache.sysml.runtime.controlprogram.context.SparkExecutionContext; import org.apache.sysml.runtime.instructions.InstructionUtils; @@ -190,7 +189,7 @@ public class WriteSPInstruction extends SPInstruction { // This case is applicable when the CSV output from transform() is written out @SuppressWarnings("unchecked") - JavaPairRDD<Long,String> rdd = (JavaPairRDD<Long, String>) ((MatrixObject) sec.getVariable(input1.getName())).getRDDHandle().getRDD(); + JavaPairRDD<Long,String> rdd = (JavaPairRDD<Long, String>) (sec.getMatrixObject(input1.getName())).getRDDHandle().getRDD(); out = rdd.values(); String sep = ",";
