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 2016/04/13 20:35:35 UTC
[1/2] incubator-systemml git commit: [SYSTEMML-570] Consolidation
buffer pool primitives matrices and frames
Repository: incubator-systemml
Updated Branches:
refs/heads/master d52283a16 -> 7528146ce
[SYSTEMML-570] Consolidation buffer pool primitives matrices and frames
This patch moves buffer pool logic from matrix object to cacheable data
in order to reuse the same logic for frame objects too. Accordingly, the
internal apis changed slightly to allow matrix/frame specialization with
minimal code duplication because it would create maintenance problems.
As a nice side effect, frame objects now also reuse the same
infrastructure for RDD and broadcast handles.
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/1fd254ff
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/1fd254ff
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/1fd254ff
Branch: refs/heads/master
Commit: 1fd254ffca005e8337daffec454ef8dbec19cf72
Parents: d52283a
Author: Matthias Boehm <mb...@us.ibm.com>
Authored: Tue Apr 12 16:36:14 2016 -0700
Committer: Matthias Boehm <mb...@us.ibm.com>
Committed: Tue Apr 12 16:36:14 2016 -0700
----------------------------------------------------------------------
.../controlprogram/caching/CacheBlock.java | 5 +
.../controlprogram/caching/CacheableData.java | 667 ++++++++++++++-
.../controlprogram/caching/FrameObject.java | 78 +-
.../controlprogram/caching/MatrixObject.java | 815 ++-----------------
.../instructions/spark/data/LineageObject.java | 35 +-
.../sysml/runtime/matrix/data/FrameBlock.java | 5 +
.../sysml/runtime/matrix/data/MatrixBlock.java | 6 +
7 files changed, 774 insertions(+), 837 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheBlock.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheBlock.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheBlock.java
index fd70bb1..2cc9471 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheBlock.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheBlock.java
@@ -49,4 +49,9 @@ public interface CacheBlock extends Writable
* @return
*/
public boolean isShallowSerialize();
+
+ /**
+ * Free unnecessarily allocated empty block.
+ */
+ public void compactEmptyBlock();
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
index 953f70b..c75820f 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
@@ -23,17 +23,28 @@ import java.io.File;
import java.io.IOException;
import java.lang.ref.SoftReference;
+import org.apache.commons.lang.mutable.MutableBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.sysml.api.DMLScript;
+import org.apache.sysml.api.DMLScript.RUNTIME_PLATFORM;
+import org.apache.sysml.conf.ConfigurationManager;
import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.runtime.DMLRuntimeException;
import org.apache.sysml.runtime.controlprogram.caching.LazyWriteBuffer.RPolicy;
import org.apache.sysml.runtime.controlprogram.parfor.util.IDSequence;
import org.apache.sysml.runtime.instructions.cp.Data;
+import org.apache.sysml.runtime.instructions.spark.data.BroadcastObject;
+import org.apache.sysml.runtime.instructions.spark.data.RDDObject;
+import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
+import org.apache.sysml.runtime.matrix.MatrixFormatMetaData;
+import org.apache.sysml.runtime.matrix.MetaData;
import org.apache.sysml.runtime.matrix.data.FileFormatProperties;
+import org.apache.sysml.runtime.matrix.data.InputInfo;
+import org.apache.sysml.runtime.matrix.data.OutputInfo;
import org.apache.sysml.runtime.util.LocalFileUtils;
+import org.apache.sysml.runtime.util.MapReduceTool;
/**
@@ -95,6 +106,13 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
public static String cacheEvictionLocalFilePath = null; //set during init
public static String cacheEvictionLocalFilePrefix = "cache";
+ /**
+ * Current state of pinned variables, required for guarded collect.
+ */
+ private static ThreadLocal<Long> sizePinned = new ThreadLocal<Long>() {
+ @Override protected Long initialValue() { return 0L; }
+ };
+
static {
_seq = new IDSequence();
}
@@ -114,6 +132,17 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
/** Container object that holds the actual data. */
protected T _data = null;
+ /**
+ * Object that holds the metadata associated with the matrix, which
+ * includes: 1) Matrix dimensions, if available 2) Number of non-zeros, if
+ * available 3) Block dimensions, if applicable 4) InputInfo -- subsequent
+ * operations that use this Matrix expect it to be in this format.
+ *
+ * When the matrix is written to HDFS (local file system, as well?), one
+ * must get the OutputInfo that matches with InputInfo stored inside _mtd.
+ */
+ protected MetaData _metaData = null;
+
/** The name of HDFS file in which the data is backed up. */
protected String _hdfsFileName = null; // file name and path
@@ -135,13 +164,20 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
*/
private boolean _dirtyFlag = false;
-
// additional private flags and meta data
private int _numReadThreads = 0; //number of threads for read from HDFS
private boolean _cleanupFlag = true; //flag if obj unpinned (cleanup enabled)
private String _varName = ""; //plan variable name
private String _cacheFileName = null; //local eviction file name
-
+ private boolean _requiresLocalWrite = false; //flag if local write for read obj
+ private boolean _isAcquireFromEmpty = false; //flag if read from status empty
+
+ //spark-specific handles
+ //note: we use the abstraction of LineageObjects for two reasons: (1) to keep track of cleanup
+ //for lazily evaluated RDDs, and (2) as abstraction for environments that do not necessarily have spark libraries available
+ private RDDObject _rddHandle = null; //RDD handle
+ private BroadcastObject _bcHandle = null; //Broadcast handle
+
/**
* Basic constructor for any cacheable data.
*
@@ -273,46 +309,374 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
_formatProps = props;
}
- // --------- ABSTRACT HIGH-LEVEL CACHE I/O OPERATIONS ----------
+ @Override
+ public void setMetaData(MetaData md) {
+ _metaData = md;
+ }
+
+ @Override
+ public MetaData getMetaData() {
+ return _metaData;
+ }
+ @Override
+ public void removeMetaData() {
+ _metaData = null;
+ }
+
/**
*
- * @return
- * @throws CacheException
*/
- public abstract T acquireRead()
+ public abstract void refreshMetaData()
throws CacheException;
-
+
/**
*
* @return
- * @throws CacheException
*/
- public abstract T acquireModify()
- throws CacheException;
+ public RDDObject getRDDHandle() {
+ return _rddHandle;
+ }
+
+ /**
+ *
+ * @param rdd
+ */
+ public void setRDDHandle( RDDObject rdd ) {
+ //cleanup potential old back reference
+ if( _rddHandle != null )
+ _rddHandle.setBackReference(null);
+ //add new rdd handle
+ _rddHandle = rdd;
+ if( _rddHandle != null )
+ rdd.setBackReference(this);
+ }
+
+ public BroadcastObject getBroadcastHandle() {
+ return _bcHandle;
+ }
+
/**
*
- * @param newData
- * @return
- * @throws CacheException
+ * @param bc
*/
- public abstract T acquireModify(T newData)
- throws CacheException;
+ public void setBroadcastHandle( BroadcastObject bc ) {
+ //cleanup potential old back reference
+ if( _bcHandle != null )
+ _bcHandle.setBackReference(null);
+
+ //add new broadcast handle
+ _bcHandle = bc;
+ if( _bcHandle != null )
+ bc.setBackReference(this);
+ }
+
+ // *********************************************
+ // *** ***
+ // *** HIGH-LEVEL METHODS THAT SPECIFY ***
+ // *** THE LOCKING AND CACHING INTERFACE ***
+ // *** ***
+ // *********************************************
/**
+ * Acquires a shared "read-only" lock, produces the reference to the cache block,
+ * restores the cache block to main memory, reads from HDFS if needed.
*
- * @throws CacheException
+ * Synchronized because there might be parallel threads (parfor local) that
+ * access the same object (in case it was created before the loop).
+ *
+ * In-Status: EMPTY, EVICTABLE, EVICTED, READ;
+ * Out-Status: READ(+1).
+ *
+ * @return
+ * @throws CacheException
*/
- public abstract void release()
- throws CacheException;
+ public synchronized T acquireRead()
+ throws CacheException
+ {
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Acquire read "+getVarName());
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+
+ if ( !isAvailableToRead() )
+ throw new CacheException ("MatrixObject not available to read.");
+
+ //get object from cache
+ if( _data == null )
+ getCache();
+
+ //read data from HDFS/RDD if required
+ //(probe data for cache_nowrite / jvm_reuse)
+ if( isEmpty(true) && _data==null )
+ {
+ try
+ {
+ if( DMLScript.STATISTICS )
+ CacheStatistics.incrementHDFSHits();
+
+ if( getRDDHandle()==null || getRDDHandle().allowsShortCircuitRead() )
+ {
+ //check filename
+ if( _hdfsFileName == null )
+ throw new CacheException("Cannot read matrix for empty filename.");
+
+ //read cacheable data from hdfs
+ _data = readBlobFromHDFS( _hdfsFileName );
+
+ //mark for initial local write despite read operation
+ _requiresLocalWrite = CACHING_WRITE_CACHE_ON_READ;
+ }
+ else
+ {
+ //read matrix from rdd (incl execute pending rdd operations)
+ MutableBoolean writeStatus = new MutableBoolean();
+ _data = readBlobFromRDD( getRDDHandle(), writeStatus );
+
+ //mark for initial local write (prevent repeated execution of rdd operations)
+ if( writeStatus.booleanValue() )
+ _requiresLocalWrite = CACHING_WRITE_CACHE_ON_READ;
+ else
+ _requiresLocalWrite = true;
+ }
+
+ setDirty(false);
+ }
+ catch (IOException e) {
+ throw new CacheException("Reading of " + _hdfsFileName + " ("+getVarName()+") failed.", e);
+ }
+
+ _isAcquireFromEmpty = true;
+ }
+ else if( DMLScript.STATISTICS )
+ {
+ if( _data!=null )
+ CacheStatistics.incrementMemHits();
+ }
+
+ //cache status maintenance
+ acquire( false, _data==null );
+ updateStatusPinned(true);
+
+ if( DMLScript.STATISTICS ){
+ long t1 = System.nanoTime();
+ CacheStatistics.incrementAcquireRTime(t1-t0);
+ }
+
+ return _data;
+ }
/**
+ * Acquires the exclusive "write" lock for a thread that wants to change cache block
+ * cell values. Produces the reference to the cache block, restores the cache block
+ * to main memory, reads from HDFS if needed.
+ *
+ * In-Status: EMPTY, EVICTABLE, EVICTED;
+ * Out-Status: MODIFY.
+ *
+ * @return
+ * @throws CacheException
+ */
+ public synchronized T acquireModify()
+ throws CacheException
+ {
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Acquire modify "+getVarName());
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+
+ if ( !isAvailableToModify() )
+ throw new CacheException("MatrixObject not available to modify.");
+
+ //get object from cache
+ if( _data == null )
+ getCache();
+
+ //read data from HDFS if required
+ if( isEmpty(true) && _data == null )
+ {
+ //check filename
+ if( _hdfsFileName == null )
+ throw new CacheException("Cannot read matrix for empty filename.");
+
+ //load data
+ try
+ {
+ _data = readBlobFromHDFS( _hdfsFileName );
+ }
+ catch (IOException e)
+ {
+ throw new CacheException("Reading of " + _hdfsFileName + " ("+getVarName()+") failed.", e);
+ }
+ }
+
+ //cache status maintenance
+ acquire( true, _data==null );
+ updateStatusPinned(true);
+ setDirty(true);
+ _isAcquireFromEmpty = false;
+
+ if( DMLScript.STATISTICS ){
+ long t1 = System.nanoTime();
+ CacheStatistics.incrementAcquireMTime(t1-t0);
+ }
+
+ return _data;
+ }
+
+ /**
+ * Acquires the exclusive "write" lock for a thread that wants to throw away the
+ * old cache block data and link up with new cache block data. Abandons the old data
+ * without reading it and sets the new data reference.
+
+ * In-Status: EMPTY, EVICTABLE, EVICTED;
+ * Out-Status: MODIFY.
+ *
+ * @param newData
+ * @return
+ * @throws CacheException
+ */
+ public synchronized T acquireModify(T newData)
+ throws CacheException
+ {
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Acquire modify newdata "+getVarName());
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+
+ if (! isAvailableToModify ())
+ throw new CacheException ("CacheableData not available to modify.");
+
+ //clear old data
+ clearData();
+
+ //cache status maintenance
+ acquire (true, false); //no need to load evicted matrix
+ setDirty(true);
+ _isAcquireFromEmpty = false;
+
+ //set references to new data
+ if (newData == null)
+ throw new CacheException("acquireModify with empty cache block.");
+ _data = newData;
+ updateStatusPinned(true);
+
+ if( DMLScript.STATISTICS ){
+ long t1 = System.nanoTime();
+ CacheStatistics.incrementAcquireMTime(t1-t0);
+ }
+
+ return _data;
+ }
+
+ /**
+ * Releases the shared ("read-only") or exclusive ("write") lock. Updates
+ * size information, last-access time, metadata, etc.
+ *
+ * Synchronized because there might be parallel threads (parfor local) that
+ * access the same object (in case it was created before the loop).
+ *
+ * In-Status: READ, MODIFY;
+ * Out-Status: READ(-1), EVICTABLE, EMPTY.
*
* @throws CacheException
*/
- public abstract void clearData()
- throws CacheException;
+ public synchronized void release()
+ throws CacheException
+ {
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Release "+getVarName());
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+
+ boolean write = false;
+ if ( isModify() )
+ {
+ //set flags for write
+ write = true;
+ setDirty(true);
+
+ //update meta data
+ refreshMetaData();
+ }
+
+ //compact empty in-memory block
+ _data.compactEmptyBlock();
+
+ //cache status maintenance (pass cacheNoWrite flag)
+ release(_isAcquireFromEmpty && !_requiresLocalWrite);
+ updateStatusPinned(false);
+
+ if( isCachingActive() //only if caching is enabled (otherwise keep everything in mem)
+ && isCached(true) //not empty and not read/modify
+ && !isBelowCachingThreshold() ) //min size for caching
+ {
+ if( write || _requiresLocalWrite )
+ {
+ //evict blob
+ String filePath = getCacheFilePathAndName();
+ try {
+ LazyWriteBuffer.writeBlock(filePath, _data);
+ }
+ catch (Exception e)
+ {
+ throw new CacheException("Eviction to local path " + filePath + " ("+getVarName()+") failed.", e);
+ }
+ _requiresLocalWrite = false;
+ }
+
+ //create cache
+ createCache();
+ _data = null;
+ }
+ else if( LOG.isTraceEnabled() ){
+ LOG.trace("Var "+getVarName()+" not subject to caching, state="+getStatusAsString());
+ }
+
+ if( DMLScript.STATISTICS ){
+ long t1 = System.nanoTime();
+ CacheStatistics.incrementReleaseTime(t1-t0);
+ }
+ }
+
+ /**
+ * Sets the cache block reference to <code>null</code>, abandons the old block.
+ * Makes the "envelope" empty. Run it to finalize the object (otherwise the
+ * evicted cache block file may remain undeleted).
+ *
+ * In-Status: EMPTY, EVICTABLE, EVICTED;
+ * Out-Status: EMPTY.
+ * @throws CacheException
+ */
+ public synchronized void clearData()
+ throws CacheException
+ {
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Clear data "+getVarName());
+
+ // check if cleanup enabled and possible
+ if( !isCleanupEnabled() )
+ return; // do nothing
+ if( !isAvailableToModify() )
+ throw new CacheException ("CacheableData (" + getDebugName() + ") not available to "
+ + "modify. Status = " + getStatusAsString() + ".");
+
+ // clear existing WB / FS representation (but prevent unnecessary probes)
+ if( !(isEmpty(true)||(_data!=null && isBelowCachingThreshold())
+ ||(_data!=null && !isCachingActive()) )) //additional condition for JMLC
+ freeEvictedBlob();
+
+ // clear the in-memory data
+ _data = null;
+ clearCache();
+
+ // clear rdd/broadcast back refs
+ if( _rddHandle != null )
+ _rddHandle.setBackReference(null);
+ if( _bcHandle != null )
+ _bcHandle.setBackReference(null);
+
+ // change object state EMPTY
+ setDirty(false);
+ setEmpty();
+ }
/**
*
@@ -323,7 +687,7 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
}
/**
- * Writes, or flushes, the matrix data to HDFS.
+ * Writes, or flushes, the cache block data to HDFS.
*
* In-Status: EMPTY, EVICTABLE, EVICTED, READ;
* Out-Status: EMPTY, EVICTABLE, EVICTED, READ.
@@ -363,16 +727,128 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
}
/**
+ * Synchronized because there might be parallel threads (parfor local) that
+ * access the same object (in case it was created before the loop).
+ * If all threads export the same data object concurrently it results in errors
+ * because they all write to the same file. Efficiency for loops and parallel threads
+ * is achieved by checking if the in-memory block is dirty.
+ *
+ * NOTE: MB: we do not use dfs copy from local (evicted) to HDFS because this would ignore
+ * the output format and most importantly would bypass reblocking during write (which effects the
+ * potential degree of parallelism). However, we copy files on HDFS if certain criteria are given.
*
* @param fName
* @param outputFormat
- * @param replication
- * @param formatProperties
* @throws CacheException
*/
- public abstract void exportData (String fName, String outputFormat, int replication, FileFormatProperties formatProperties)
- throws CacheException;
+ public synchronized void exportData (String fName, String outputFormat, int replication, FileFormatProperties formatProperties)
+ throws CacheException
+ {
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Export data "+getVarName()+" "+fName);
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+
+ //prevent concurrent modifications
+ if ( !isAvailableToRead() )
+ throw new CacheException ("MatrixObject not available to read.");
+ LOG.trace("Exporting " + this.getDebugName() + " to " + fName + " in format " + outputFormat);
+
+ boolean pWrite = false; // !fName.equals(_hdfsFileName); //persistent write flag
+ if ( fName.equals(_hdfsFileName) ) {
+ setHDFSFileExists(true);
+ pWrite = false;
+ }
+ else {
+ pWrite = true; // i.e., export is called from "write" instruction
+ }
+
+ //actual export (note: no direct transfer of local copy in order to ensure blocking (and hence, parallelism))
+ if( isDirty() || //use dirty for skipping parallel exports
+ (pWrite && !isEqualOutputFormat(outputFormat)) )
+ {
+ // CASE 1: dirty in-mem matrix or pWrite w/ different format (write matrix to fname; load into memory if evicted)
+ // a) get the matrix
+ if( isEmpty(true) )
+ {
+ //read data from HDFS if required (never read before), this applies only to pWrite w/ different output formats
+ //note: for large rdd outputs, we compile dedicated writespinstructions (no need to handle this here)
+ try
+ {
+ if( getRDDHandle()==null || getRDDHandle().allowsShortCircuitRead() )
+ _data = readBlobFromHDFS( _hdfsFileName );
+ else
+ _data = readBlobFromRDD( getRDDHandle(), new MutableBoolean() );
+ setDirty(false);
+ }
+ catch (IOException e)
+ {
+ throw new CacheException("Reading of " + _hdfsFileName + " ("+getVarName()+") failed.", e);
+ }
+ }
+ //get object from cache
+ if( _data == null )
+ getCache();
+ acquire( false, _data==null ); //incl. read matrix if evicted
+
+ // b) write the matrix
+ try
+ {
+ writeMetaData( fName, outputFormat, formatProperties );
+ writeBlobToHDFS( fName, outputFormat, replication, formatProperties );
+ if ( !pWrite )
+ setDirty(false);
+ }
+ catch (Exception e)
+ {
+ throw new CacheException ("Export to " + fName + " failed.", e);
+ }
+ finally
+ {
+ release();
+ }
+ }
+ else if( pWrite ) // pwrite with same output format
+ {
+ //CASE 2: matrix already in same format but different file on hdfs (copy matrix to fname)
+ try
+ {
+ MapReduceTool.deleteFileIfExistOnHDFS(fName);
+ MapReduceTool.deleteFileIfExistOnHDFS(fName+".mtd");
+ if( getRDDHandle()==null || getRDDHandle().allowsShortCircuitRead() )
+ MapReduceTool.copyFileOnHDFS( _hdfsFileName, fName );
+ else //write might trigger rdd operations and nnz maintenance
+ writeBlobFromRDDtoHDFS(getRDDHandle(), fName, outputFormat);
+ writeMetaData( fName, outputFormat, formatProperties );
+ }
+ catch (Exception e) {
+ throw new CacheException ("Export to " + fName + " failed.", e);
+ }
+ }
+ else if( getRDDHandle()!=null && //pending rdd operation
+ !getRDDHandle().allowsShortCircuitRead() )
+ {
+ //CASE 3: pending rdd operation (other than checkpoints)
+ try
+ {
+ writeBlobFromRDDtoHDFS(getRDDHandle(), fName, outputFormat);
+ writeMetaData( fName, outputFormat, formatProperties );
+ }
+ catch (Exception e) {
+ throw new CacheException ("Export to " + fName + " failed.", e);
+ }
+ }
+ else
+ {
+ //CASE 4: data already in hdfs (do nothing, no need for export)
+ LOG.trace(this.getDebugName() + ": Skip export to hdfs since data already exists.");
+ }
+
+ if( DMLScript.STATISTICS ){
+ long t1 = System.nanoTime();
+ CacheStatistics.incrementExportTime(t1-t0);
+ }
+ }
// --------- ABSTRACT LOW-LEVEL CACHE I/O OPERATIONS ----------
@@ -391,7 +867,7 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
/**
* Low-level cache I/O method that physically restores the data blob to
- * main memory. Must be defined by a subclass, never called by users.
+ * main memory. Must be defined by a subclass, never called by users.
*
* @throws CacheException
*/
@@ -466,6 +942,122 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
_hdfsFileName.substring (_hdfsFileName.length() - maxLength + 3)));
return getVarName() + " " + debugNameEnding;
}
+
+ /**
+ *
+ * @param fname
+ * @return
+ * @throws IOException
+ */
+ protected T readBlobFromHDFS(String fname) throws IOException {
+ MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
+ MatrixCharacteristics mc = iimd.getMatrixCharacteristics();
+ return readBlobFromHDFS(fname, mc.getRows(), mc.getCols());
+ }
+
+ /**
+ *
+ * @param fname
+ * @param rlen
+ * @param clen
+ * @return
+ * @throws IOException
+ */
+ protected abstract T readBlobFromHDFS(String fname, long rlen, long clen)
+ throws IOException;
+
+ /**
+ *
+ * @param rdd
+ * @param status
+ * @return
+ * @throws IOException
+ */
+ protected abstract T readBlobFromRDD(RDDObject rdd, MutableBoolean status)
+ throws IOException;
+
+ /**
+ *
+ * @param fname
+ * @param ofmt
+ * @param rep
+ * @param fprop
+ * @throws IOException
+ * @throws DMLRuntimeException
+ */
+ protected abstract void writeBlobToHDFS(String fname, String ofmt, int rep, FileFormatProperties fprop)
+ throws IOException, DMLRuntimeException;
+
+ /**
+ *
+ * @param rdd
+ * @param fname
+ * @param ofmt
+ * @throws IOException
+ * @throws DMLRuntimeException
+ */
+ protected abstract void writeBlobFromRDDtoHDFS(RDDObject rdd, String fname, String ofmt)
+ throws IOException, DMLRuntimeException;
+
+
+ /**
+ *
+ * @param filePathAndName
+ * @param outputFormat
+ * @param formatProperties
+ * @throws DMLRuntimeException
+ * @throws IOException
+ */
+ protected void writeMetaData (String filePathAndName, String outputFormat, FileFormatProperties formatProperties)
+ throws DMLRuntimeException, IOException
+ {
+ MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
+
+ if (iimd == null)
+ throw new DMLRuntimeException("Unexpected error while writing mtd file (" + filePathAndName + ") -- metadata is null.");
+
+ // Write the matrix to HDFS in requested format
+ OutputInfo oinfo = (outputFormat != null ? OutputInfo.stringToOutputInfo (outputFormat)
+ : InputInfo.getMatchingOutputInfo (iimd.getInputInfo ()));
+
+ if ( oinfo != OutputInfo.MatrixMarketOutputInfo ) {
+ // Get the dimension information from the metadata stored within MatrixObject
+ MatrixCharacteristics mc = iimd.getMatrixCharacteristics ();
+
+ // when outputFormat is binaryblock, make sure that matrixCharacteristics has correct blocking dimensions
+ // note: this is only required if singlenode (due to binarycell default)
+ if ( oinfo == OutputInfo.BinaryBlockOutputInfo && DMLScript.rtplatform == RUNTIME_PLATFORM.SINGLE_NODE &&
+ (mc.getRowsPerBlock() != ConfigurationManager.getBlocksize() || mc.getColsPerBlock() != ConfigurationManager.getBlocksize()) )
+ {
+ mc = new MatrixCharacteristics(mc.getRows(), mc.getCols(), ConfigurationManager.getBlocksize(), ConfigurationManager.getBlocksize(), mc.getNonZeros());
+ }
+ MapReduceTool.writeMetaDataFile (filePathAndName + ".mtd", valueType, mc, oinfo, formatProperties);
+ }
+ }
+
+ /**
+ *
+ * @param outputFormat
+ * @return
+ */
+ protected boolean isEqualOutputFormat( String outputFormat )
+ {
+ boolean ret = true;
+ if( outputFormat != null ) {
+ try {
+ MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
+ OutputInfo oi1 = InputInfo.getMatchingOutputInfo( iimd.getInputInfo() );
+ OutputInfo oi2 = OutputInfo.stringToOutputInfo( outputFormat );
+ if( oi1 != oi2 )
+ ret = false;
+ }
+ catch(Exception ex) {
+ ret = false;
+ }
+ }
+
+ return ret;
+ }
// ------------- IMPLEMENTED CACHE LOGIC METHODS --------------
@@ -597,6 +1189,10 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
return (_cacheStatus == CacheStatus.CACHED);
}
+ public void setEmptyStatus() {
+ setEmpty();
+ }
+
protected boolean isEmpty(boolean inclCachedNoWrite) {
if( inclCachedNoWrite )
return (_cacheStatus == CacheStatus.EMPTY || _cacheStatus == CacheStatus.CACHED_NOWRITE);
@@ -684,6 +1280,25 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
}
}
+ /**
+ *
+ * @param add
+ */
+ protected void updateStatusPinned(boolean add) {
+ if( _data != null ) { //data should never be null
+ long size = sizePinned.get();
+ size += (add ? 1 : -1) * _data.getInMemorySize();
+ sizePinned.set( Math.max(size,0) );
+ }
+ }
+
+ /**
+ *
+ * @return
+ */
+ protected long getPinnedSize() {
+ return sizePinned.get();
+ }
// --------- STATIC CACHE INIT/CLEANUP OPERATIONS ----------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
index 0a014ed..f83e7c1 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
@@ -22,8 +22,11 @@ package org.apache.sysml.runtime.controlprogram.caching;
import java.io.IOException;
+import org.apache.commons.lang.mutable.MutableBoolean;
import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.parser.Expression.ValueType;
+import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.instructions.spark.data.RDDObject;
import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
import org.apache.sysml.runtime.matrix.MatrixDimensionsMetaData;
import org.apache.sysml.runtime.matrix.MetaData;
@@ -34,9 +37,6 @@ public class FrameObject extends CacheableData<FrameBlock>
{
private static final long serialVersionUID = 1755082174281927785L;
- /** Holds meta data on frame characteristics (required for nrow etc)*/
- private MetaData _metaData = null;
-
/**
*
*/
@@ -72,22 +72,8 @@ public class FrameObject extends CacheableData<FrameBlock>
public FrameObject(FrameObject fo) {
super(fo);
}
-
- @Override
- public void setMetaData(MetaData md) {
- _metaData = md;
- }
-
- @Override
- public MetaData getMetaData() {
- return _metaData;
- }
@Override
- public void removeMetaData() {
- _metaData = null;
- }
-
public void refreshMetaData()
throws CacheException
{
@@ -98,67 +84,37 @@ public class FrameObject extends CacheableData<FrameBlock>
mc.setDimension( _data.getNumRows(),_data.getNumColumns() );
}
- ////////////////////////////////////
- // high-level cache API (pseudo-integrated)
-
@Override
- public FrameBlock acquireRead()
- throws CacheException
- {
- //read in=memory frame if necessary
- if( _data == null ) {
- // TODO @Arvind: please integrate readers here for now
- }
-
- return _data;
- }
-
- @Override
- public FrameBlock acquireModify()
- throws CacheException
- {
- return _data;
+ protected FrameBlock readBlobFromCache(String fname) throws IOException {
+ return (FrameBlock)LazyWriteBuffer.readBlock(fname, false);
}
@Override
- public FrameBlock acquireModify(FrameBlock newData)
- throws CacheException
+ protected FrameBlock readBlobFromHDFS(String fname, long rlen, long clen)
+ throws IOException
{
- //set data and update meta data
- _data = newData;
- refreshMetaData();
-
- return _data;
+ throw new IOException("Not implemented yet.");
}
@Override
- public void release()
- throws CacheException
+ protected FrameBlock readBlobFromRDD(RDDObject rdd, MutableBoolean status)
+ throws IOException
{
- //do nothing
+ throw new IOException("Not implemented yet.");
}
@Override
- public void clearData()
- throws CacheException
+ protected void writeBlobToHDFS(String fname, String ofmt, int rep, FileFormatProperties fprop)
+ throws IOException, DMLRuntimeException
{
- if( isCleanupEnabled() )
- _data = null;
+ throw new IOException("Not implemented yet.");
}
@Override
- public void exportData(String fName, String outputFormat, int replication, FileFormatProperties formatProperties)
- throws CacheException
+ protected void writeBlobFromRDDtoHDFS(RDDObject rdd, String fname, String ofmt)
+ throws IOException, DMLRuntimeException
{
- // TODO @Arvind: please integrate writers here for now
- }
-
- ////////////////////////////////////
- // low-level cache API
-
- @Override
- protected FrameBlock readBlobFromCache(String fname) throws IOException {
- return (FrameBlock)LazyWriteBuffer.readBlock(fname, false);
+ throw new IOException("Not implemented yet.");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/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 f946633..cb27ee4 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
@@ -33,7 +33,6 @@ import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.runtime.DMLRuntimeException;
import org.apache.sysml.runtime.controlprogram.ParForProgramBlock.PDataPartitionFormat;
import org.apache.sysml.runtime.controlprogram.context.SparkExecutionContext;
-import org.apache.sysml.runtime.instructions.spark.data.BroadcastObject;
import org.apache.sysml.runtime.instructions.spark.data.RDDObject;
import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
import org.apache.sysml.runtime.matrix.MatrixDimensionsMetaData;
@@ -62,39 +61,11 @@ import org.apache.sysml.runtime.util.MapReduceTool;
public class MatrixObject extends CacheableData<MatrixBlock>
{
private static final long serialVersionUID = 6374712373206495637L;
-
- /**
- * Current state of pinned variables, required for guarded collect.
- */
- private static ThreadLocal<Long> sizePinned = new ThreadLocal<Long>() {
- @Override protected Long initialValue() { return 0L; }
- };
-
- /**
- * Object that holds the metadata associated with the matrix, which
- * includes: 1) Matrix dimensions, if available 2) Number of non-zeros, if
- * available 3) Block dimensions, if applicable 4) InputInfo -- subsequent
- * operations that use this Matrix expect it to be in this format.
- *
- * When the matrix is written to HDFS (local file system, as well?), one
- * must get the OutputInfo that matches with InputInfo stored inside _mtd.
- */
- private MetaData _metaData = null;
- //additional names and flags
- private boolean _requiresLocalWrite = false; //flag if local write for read obj
- private boolean _isAcquireFromEmpty = false; //flag if read from status empty
+ //additional matrix-specific flags
private boolean _updateInPlaceFlag = false; //flag if in-place update
- //spark-specific handles
- //note: we use the abstraction of LineageObjects for two reasons: (1) to keep track of cleanup
- //for lazily evaluated RDDs, and (2) as abstraction for environments that do not necessarily have spark libraries available
- private RDDObject _rddHandle = null; //RDD handle
- private BroadcastObject _bcHandle = null; //Broadcast handle
-
- /**
- * Information relevant to partitioned matrices.
- */
+ //information relevant to partitioned matrices.
private boolean _partitioned = false; //indicates if obj partitioned
private PDataPartitionFormat _partitionFormat = null; //indicates how obj partitioned
private int _partitionSize = -1; //indicates n for BLOCKWISE_N
@@ -104,20 +75,17 @@ public class MatrixObject extends CacheableData<MatrixBlock>
/**
* Constructor that takes only the HDFS filename.
*/
- public MatrixObject (ValueType vt, String file)
- {
+ public MatrixObject (ValueType vt, String file) {
this (vt, file, null); //HDFS file path
}
/**
* Constructor that takes both HDFS filename and associated metadata.
*/
- public MatrixObject( ValueType vt, String file, MetaData mtd )
- {
+ public MatrixObject( ValueType vt, String file, MetaData mtd ) {
super (DataType.MATRIX, vt);
_metaData = mtd;
- _hdfsFileName = file;
-
+ _hdfsFileName = file;
_cache = null;
_data = null;
}
@@ -143,19 +111,21 @@ public class MatrixObject extends CacheableData<MatrixBlock>
_partitionCacheName = mo._partitionCacheName;
}
- @Override
- public void setMetaData(MetaData md) {
- _metaData = md;
+
+ /**
+ *
+ * @param flag
+ */
+ public void enableUpdateInPlace(boolean flag) {
+ _updateInPlaceFlag = flag;
}
- @Override
- public MetaData getMetaData() {
- return _metaData;
- }
-
- @Override
- public void removeMetaData() {
- _metaData = null;
+ /**
+ *
+ * @return
+ */
+ public boolean isUpdateInPlaceEnabled() {
+ return _updateInPlaceFlag;
}
@Override
@@ -167,6 +137,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
* Make the matrix metadata consistent with the in-memory matrix data
* @throws CacheException
*/
+ @Override
public void refreshMetaData()
throws CacheException
{
@@ -184,8 +155,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public long getNumRows ()
- {
+ public long getNumRows() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
MatrixCharacteristics mc = meta.getMatrixCharacteristics();
return mc.getRows ();
@@ -195,8 +165,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public long getNumColumns()
- {
+ public long getNumColumns() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
MatrixCharacteristics mc = meta.getMatrixCharacteristics();
return mc.getCols ();
@@ -206,8 +175,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public long getNumRowsPerBlock()
- {
+ public long getNumRowsPerBlock() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
MatrixCharacteristics mc = meta.getMatrixCharacteristics();
return mc.getRowsPerBlock();
@@ -217,8 +185,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public long getNumColumnsPerBlock()
- {
+ public long getNumColumnsPerBlock() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
MatrixCharacteristics mc = meta.getMatrixCharacteristics();
return mc.getColsPerBlock();
@@ -228,8 +195,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public long getNnz()
- {
+ public long getNnz() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
MatrixCharacteristics mc = meta.getMatrixCharacteristics();
return mc.getNonZeros();
@@ -239,11 +205,9 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public double getSparsity()
- {
+ public double getSparsity() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
-
+ MatrixCharacteristics mc = meta.getMatrixCharacteristics();
return ((double)mc.getNonZeros())/mc.getRows()/mc.getCols();
}
@@ -251,8 +215,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
*
* @return
*/
- public MatrixCharacteristics getMatrixCharacteristics()
- {
+ public MatrixCharacteristics getMatrixCharacteristics() {
MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
return meta.getMatrixCharacteristics();
}
@@ -297,480 +260,6 @@ public class MatrixObject extends CacheableData<MatrixBlock>
return str.toString();
}
-
- public RDDObject getRDDHandle()
- {
- return _rddHandle;
- }
-
- public void setRDDHandle( RDDObject rdd )
- {
- //cleanup potential old back reference
- if( _rddHandle != null )
- _rddHandle.setBackReference(null);
-
- //add new rdd handle
- _rddHandle = rdd;
- if( _rddHandle != null )
- rdd.setBackReference(this);
- }
-
- public BroadcastObject getBroadcastHandle()
- {
- return _bcHandle;
- }
-
- public void setBroadcastHandle( BroadcastObject bc )
- {
- //cleanup potential old back reference
- if( _bcHandle != null )
- _bcHandle.setBackReference(null);
-
- //add new broadcast handle
- _bcHandle = bc;
- if( _bcHandle != null )
- bc.setBackReference(this);
- }
-
-
- // *********************************************
- // *** ***
- // *** HIGH-LEVEL METHODS THAT SPECIFY ***
- // *** THE LOCKING AND CACHING INTERFACE ***
- // *** ***
- // *********************************************
-
-
- /**
- * Acquires a shared "read-only" lock, produces the reference to the matrix data,
- * restores the matrix to main memory, reads from HDFS if needed.
- *
- * Synchronized because there might be parallel threads (parfor local) that
- * access the same MatrixObjectNew object (in case it was created before the loop).
- *
- * In-Status: EMPTY, EVICTABLE, EVICTED, READ;
- * Out-Status: READ(+1).
- *
- * @return the matrix data reference
- * @throws CacheException
- */
- @Override
- public synchronized MatrixBlock acquireRead()
- throws CacheException
- {
- if( LOG.isTraceEnabled() )
- LOG.trace("Acquire read "+getVarName());
- long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
-
- if ( !isAvailableToRead() )
- throw new CacheException ("MatrixObject not available to read.");
-
- //get object from cache
- if( _data == null )
- getCache();
-
- //read data from HDFS/RDD if required
- //(probe data for cache_nowrite / jvm_reuse)
- if( isEmpty(true) && _data==null )
- {
- try
- {
- if( DMLScript.STATISTICS )
- CacheStatistics.incrementHDFSHits();
-
- if( getRDDHandle()==null || getRDDHandle().allowsShortCircuitRead() )
- {
- //check filename
- if( _hdfsFileName == null )
- throw new CacheException("Cannot read matrix for empty filename.");
-
- //read matrix from hdfs
- _data = readMatrixFromHDFS( _hdfsFileName );
-
- //mark for initial local write despite read operation
- _requiresLocalWrite = CACHING_WRITE_CACHE_ON_READ;
- }
- else
- {
- //read matrix from rdd (incl execute pending rdd operations)
- MutableBoolean writeStatus = new MutableBoolean();
- _data = readMatrixFromRDD( getRDDHandle(), writeStatus );
-
- //mark for initial local write (prevent repeated execution of rdd operations)
- if( writeStatus.booleanValue() )
- _requiresLocalWrite = CACHING_WRITE_CACHE_ON_READ;
- else
- _requiresLocalWrite = true;
- }
-
- setDirty(false);
- }
- catch (IOException e)
- {
- throw new CacheException("Reading of " + _hdfsFileName + " ("+getVarName()+") failed.", e);
- }
-
- _isAcquireFromEmpty = true;
- }
- else if( DMLScript.STATISTICS )
- {
- if( _data!=null )
- CacheStatistics.incrementMemHits();
- }
-
- //cache status maintenance
- super.acquire( false, _data==null );
- updateStatusPinned(true);
-
- if( DMLScript.STATISTICS ){
- long t1 = System.nanoTime();
- CacheStatistics.incrementAcquireRTime(t1-t0);
- }
-
- return _data;
- }
-
- /**
- * Acquires the exclusive "write" lock for a thread that wants to change matrix
- * cell values. Produces the reference to the matrix data, restores the matrix
- * to main memory, reads from HDFS if needed.
- *
- * In-Status: EMPTY, EVICTABLE, EVICTED;
- * Out-Status: MODIFY.
- *
- * @return the matrix data reference
- * @throws CacheException
- */
- @Override
- public synchronized MatrixBlock acquireModify()
- throws CacheException
- {
- if( LOG.isTraceEnabled() )
- LOG.trace("Acquire modify "+getVarName());
- long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
-
- if ( !isAvailableToModify() )
- throw new CacheException("MatrixObject not available to modify.");
-
- //get object from cache
- if( _data == null )
- getCache();
-
- //read data from HDFS if required
- if( isEmpty(true) && _data == null )
- {
- //check filename
- if( _hdfsFileName == null )
- throw new CacheException("Cannot read matrix for empty filename.");
-
- //load data
- try
- {
- _data = readMatrixFromHDFS( _hdfsFileName );
- }
- catch (IOException e)
- {
- throw new CacheException("Reading of " + _hdfsFileName + " ("+getVarName()+") failed.", e);
- }
- }
-
- //cache status maintenance
- super.acquire( true, _data==null );
- updateStatusPinned(true);
- setDirty(true);
- _isAcquireFromEmpty = false;
-
- if( DMLScript.STATISTICS ){
- long t1 = System.nanoTime();
- CacheStatistics.incrementAcquireMTime(t1-t0);
- }
-
- return _data;
- }
-
- /**
- * Acquires the exclusive "write" lock for a thread that wants to throw away the
- * old matrix data and link up with new matrix data. Abandons the old matrix data
- * without reading it. Sets the new matrix data reference.
-
- * In-Status: EMPTY, EVICTABLE, EVICTED;
- * Out-Status: MODIFY.
- *
- * @param newData : the new matrix data reference
- * @return the matrix data reference, which is the same as the argument
- * @throws CacheException
- */
- @Override
- public synchronized MatrixBlock acquireModify(MatrixBlock newData)
- throws CacheException
- {
- if( LOG.isTraceEnabled() )
- LOG.trace("Acquire modify newdata "+getVarName());
- long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
-
- if (! isAvailableToModify ())
- throw new CacheException ("MatrixObject not available to modify.");
-
- //clear old data
- clearData();
-
- //cache status maintenance
- super.acquire (true, false); //no need to load evicted matrix
- setDirty(true);
- _isAcquireFromEmpty = false;
-
- //set references to new data
- if (newData == null)
- throw new CacheException("acquireModify with empty matrix block.");
- _data = newData;
- updateStatusPinned(true);
-
- if( DMLScript.STATISTICS ){
- long t1 = System.nanoTime();
- CacheStatistics.incrementAcquireMTime(t1-t0);
- }
-
- return _data;
- }
-
- /**
- * Releases the shared ("read-only") or exclusive ("write") lock. Updates
- * the matrix size, last-access time, metadata, etc.
- *
- * Synchronized because there might be parallel threads (parfor local) that
- * access the same MatrixObjectNew object (in case it was created before the loop).
- *
- * In-Status: READ, MODIFY;
- * Out-Status: READ(-1), EVICTABLE, EMPTY.
- *
- * @throws CacheException
- */
- @Override
- public synchronized void release()
- throws CacheException
- {
- if( LOG.isTraceEnabled() )
- LOG.trace("Release "+getVarName());
- long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
-
- boolean write = false;
- if ( isModify() )
- {
- //set flags for write
- write = true;
- setDirty(true);
-
- //update meta data
- refreshMetaData();
- }
-
- //compact empty in-memory block
- if( _data.isEmptyBlock(false) && _data.isAllocated() )
- _data.cleanupBlock(true, true);
-
- //cache status maintenance (pass cacheNoWrite flag)
- super.release(_isAcquireFromEmpty && !_requiresLocalWrite);
- updateStatusPinned(false);
-
- if( isCachingActive() //only if caching is enabled (otherwise keep everything in mem)
- && isCached(true) //not empty and not read/modify
- && !isUpdateInPlaceEnabled() //pinned result variable
- && !isBelowCachingThreshold() ) //min size for caching
- {
- if( write || _requiresLocalWrite )
- {
- //evict blob
- String filePath = getCacheFilePathAndName();
- try {
- LazyWriteBuffer.writeBlock(filePath, _data);
- }
- catch (Exception e)
- {
- throw new CacheException("Eviction to local path " + filePath + " ("+getVarName()+") failed.", e);
- }
- _requiresLocalWrite = false;
- }
-
- //create cache
- createCache();
- _data = null;
- }
- else if( LOG.isTraceEnabled() ){
- LOG.trace("Var "+getVarName()+" not subject to caching: rows="+_data.getNumRows()+", cols="+_data.getNumColumns()+", state="+getStatusAsString());
- }
-
- if( DMLScript.STATISTICS ){
- long t1 = System.nanoTime();
- CacheStatistics.incrementReleaseTime(t1-t0);
- }
- }
-
- /**
- * Sets the matrix data reference to <code>null</code>, abandons the old matrix.
- * Makes the "envelope" empty. Run it to finalize the matrix (otherwise the
- * evicted matrix file may remain undeleted).
- *
- * In-Status: EMPTY, EVICTABLE, EVICTED;
- * Out-Status: EMPTY.
- * @throws CacheException
- */
- @Override
- public synchronized void clearData()
- throws CacheException
- {
- if( LOG.isTraceEnabled() )
- LOG.trace("Clear data "+getVarName());
-
- // check if cleanup enabled and possible
- if( !isCleanupEnabled() )
- return; // do nothing
- if( !isAvailableToModify() )
- throw new CacheException ("MatrixObject (" + this.getDebugName() + ") not available to modify. Status = " + this.getStatusAsString() + ".");
-
- // clear existing WB / FS representation (but prevent unnecessary probes)
- if( !(isEmpty(true)||(_data!=null && isBelowCachingThreshold())
- ||(_data!=null && !isCachingActive()) )) //additional condition for JMLC
- freeEvictedBlob();
-
- // clear the in-memory data
- _data = null;
- clearCache();
-
- // clear rdd/broadcast back refs
- if( _rddHandle != null )
- _rddHandle.setBackReference(null);
- if( _bcHandle != null )
- _bcHandle.setBackReference(null);
-
- // change object state EMPTY
- setDirty(false);
- setEmpty();
- }
-
- /**
- * Synchronized because there might be parallel threads (parfor local) that
- * access the same MatrixObjectNew object (in case it was created before the loop).
- * If all threads export the same data object concurrently it results in errors
- * because they all write to the same file. Efficiency for loops and parallel threads
- * is achieved by checking if the in-memory matrix block is dirty.
- *
- * NOTE: MB: we do not use dfs copy from local (evicted) to HDFS because this would ignore
- * the output format and most importantly would bypass reblocking during write (which effects the
- * potential degree of parallelism). However, we copy files on HDFS if certain criteria are given.
- *
- * @param fName
- * @param outputFormat
- * @throws CacheException
- */
- @Override
- public synchronized void exportData (String fName, String outputFormat, int replication, FileFormatProperties formatProperties)
- throws CacheException
- {
- if( LOG.isTraceEnabled() )
- LOG.trace("Export data "+getVarName()+" "+fName);
- long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
-
- //prevent concurrent modifications
- if ( !isAvailableToRead() )
- throw new CacheException ("MatrixObject not available to read.");
-
- LOG.trace("Exporting " + this.getDebugName() + " to " + fName + " in format " + outputFormat);
-
- boolean pWrite = false; // !fName.equals(_hdfsFileName); //persistent write flag
- if ( fName.equals(_hdfsFileName) ) {
- setHDFSFileExists(true);
- pWrite = false;
- }
- else {
- pWrite = true; // i.e., export is called from "write" instruction
- }
-
- //actual export (note: no direct transfer of local copy in order to ensure blocking (and hence, parallelism))
- if( isDirty() || //use dirty for skipping parallel exports
- (pWrite && !isEqualOutputFormat(outputFormat)) )
- {
- // CASE 1: dirty in-mem matrix or pWrite w/ different format (write matrix to fname; load into memory if evicted)
- // a) get the matrix
- if( isEmpty(true) )
- {
- //read data from HDFS if required (never read before), this applies only to pWrite w/ different output formats
- //note: for large rdd outputs, we compile dedicated writespinstructions (no need to handle this here)
- try
- {
- if( getRDDHandle()==null || getRDDHandle().allowsShortCircuitRead() )
- _data = readMatrixFromHDFS( _hdfsFileName );
- else
- _data = readMatrixFromRDD( getRDDHandle(), new MutableBoolean() );
- setDirty(false);
- }
- catch (IOException e)
- {
- throw new CacheException("Reading of " + _hdfsFileName + " ("+getVarName()+") failed.", e);
- }
- }
- //get object from cache
- if( _data == null )
- getCache();
- super.acquire( false, _data==null ); //incl. read matrix if evicted
-
- // b) write the matrix
- try
- {
- writeMetaData( fName, outputFormat, formatProperties );
- writeMatrixToHDFS( fName, outputFormat, replication, formatProperties );
- if ( !pWrite )
- setDirty(false);
- }
- catch (Exception e)
- {
- throw new CacheException ("Export to " + fName + " failed.", e);
- }
- finally
- {
- release();
- }
- }
- else if( pWrite ) // pwrite with same output format
- {
- //CASE 2: matrix already in same format but different file on hdfs (copy matrix to fname)
- try
- {
- MapReduceTool.deleteFileIfExistOnHDFS(fName);
- MapReduceTool.deleteFileIfExistOnHDFS(fName+".mtd");
- if( getRDDHandle()==null || getRDDHandle().allowsShortCircuitRead() )
- MapReduceTool.copyFileOnHDFS( _hdfsFileName, fName );
- else //write might trigger rdd operations and nnz maintenance
- writeMatrixFromRDDtoHDFS(getRDDHandle(), fName, outputFormat);
- writeMetaData( fName, outputFormat, formatProperties );
- }
- catch (Exception e) {
- throw new CacheException ("Export to " + fName + " failed.", e);
- }
- }
- else if( getRDDHandle()!=null && //pending rdd operation
- !getRDDHandle().allowsShortCircuitRead() )
- {
- //CASE 3: pending rdd operation (other than checkpoints)
- try
- {
- writeMatrixFromRDDtoHDFS(getRDDHandle(), fName, outputFormat);
- writeMetaData( fName, outputFormat, formatProperties );
- }
- catch (Exception e) {
- throw new CacheException ("Export to " + fName + " failed.", e);
- }
- }
- else
- {
- //CASE 4: data already in hdfs (do nothing, no need for export)
- LOG.trace(this.getDebugName() + ": Skip export to hdfs since data already exists.");
- }
-
- if( DMLScript.STATISTICS ){
- long t1 = System.nanoTime();
- CacheStatistics.incrementExportTime(t1-t0);
- }
- }
/**
*
@@ -789,7 +278,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
//ensure input file is persistent on hdfs (pending RDD operations),
//file might have been written during export or collect via write/read
if( getRDDHandle() != null && !MapReduceTool.existsFileOnHDFS(_hdfsFileName) ) {
- writeMatrixFromRDDtoHDFS(getRDDHandle(), _hdfsFileName, outputFormat);
+ writeBlobFromRDDtoHDFS(getRDDHandle(), _hdfsFileName, outputFormat);
}
//export or rename to target file on hdfs
@@ -944,7 +433,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
//read the
if( MapReduceTool.existsFileOnHDFS(fname) )
- mb = readMatrixFromHDFS( fname, rows, cols );
+ mb = readBlobFromHDFS( fname, rows, cols );
else
{
mb = new MatrixBlock((int)rows, (int)cols, true);
@@ -1039,30 +528,44 @@ public class MatrixObject extends CacheableData<MatrixBlock>
// *********************************************
@Override
+ protected boolean isBelowCachingThreshold() {
+ return super.isBelowCachingThreshold()
+ || isUpdateInPlaceEnabled(); //pinned result variable
+ }
+
+ @Override
protected MatrixBlock readBlobFromCache(String fname) throws IOException {
return (MatrixBlock)LazyWriteBuffer.readBlock(fname, true);
}
-
- // *******************************************
- // *** ***
- // *** LOW-LEVEL PRIVATE METHODS ***
- // *** FOR MATRIX I/O ***
- // *** ***
- // *******************************************
-
- /**
- *
- * @param filePathAndName
- * @return
- * @throws IOException
- */
- private MatrixBlock readMatrixFromHDFS(String filePathAndName)
+
+ @Override
+ protected MatrixBlock readBlobFromHDFS(String fname, long rlen, long clen)
throws IOException
{
MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
MatrixCharacteristics mc = iimd.getMatrixCharacteristics();
- return readMatrixFromHDFS( filePathAndName, mc.getRows(), mc.getCols() );
+ long begin = 0;
+
+ if( LOG.isTraceEnabled() ) {
+ LOG.trace("Reading matrix from HDFS... " + getVarName() + " Path: " + fname
+ + ", dimensions: [" + mc.getRows() + ", " + mc.getCols() + ", " + mc.getNonZeros() + "]");
+ begin = System.currentTimeMillis();
+ }
+
+ double sparsity = ( mc.getNonZeros() >= 0 ? ((double)mc.getNonZeros())/(mc.getRows()*mc.getCols()) : 1.0d) ;
+ MatrixBlock newData = DataConverter.readMatrixFromHDFS(fname, iimd.getInputInfo(), rlen, clen,
+ mc.getRowsPerBlock(), mc.getColsPerBlock(), sparsity, getFileFormatProperties());
+
+ //sanity check correct output
+ if( newData == null ) {
+ throw new IOException("Unable to load matrix from file: "+fname);
+ }
+
+ if( LOG.isTraceEnabled() )
+ LOG.trace("Reading Completed: " + (System.currentTimeMillis()-begin) + " msec.");
+
+ return newData;
}
/**
@@ -1071,7 +574,8 @@ public class MatrixObject extends CacheableData<MatrixBlock>
* @return
* @throws IOException
*/
- private MatrixBlock readMatrixFromRDD(RDDObject rdd, MutableBoolean writeStatus)
+ @Override
+ protected MatrixBlock readBlobFromRDD(RDDObject rdd, MutableBoolean writeStatus)
throws IOException
{
//note: the read of a matrix block from an RDD might trigger
@@ -1101,7 +605,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
//guarded rdd collect
if( ii == InputInfo.BinaryBlockInputInfo && //guarded collect not for binary cell
- !OptimizerUtils.checkSparkCollectMemoryBudget(rlen, clen, brlen, bclen, nnz, sizePinned.get()) ) {
+ !OptimizerUtils.checkSparkCollectMemoryBudget(rlen, clen, brlen, bclen, nnz, getPinnedSize()) ) {
//write RDD to hdfs and read to prevent invalid collect mem consumption
//note: lazy, partition-at-a-time collect (toLocalIterator) was significantly slower
if( !MapReduceTool.existsFileOnHDFS(_hdfsFileName) ) { //prevent overwrite existing file
@@ -1110,7 +614,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
((RDDObject)rdd).setHDFSFile(true); //mark rdd as hdfs file (for restore)
writeStatus.setValue(true); //mark for no cache-write on read
}
- mb = readMatrixFromHDFS(_hdfsFileName);
+ mb = readBlobFromHDFS(_hdfsFileName);
}
else if( ii == InputInfo.BinaryCellInputInfo ) {
//collect matrix block from binary block RDD
@@ -1134,76 +638,19 @@ public class MatrixObject extends CacheableData<MatrixBlock>
}
/**
- *
- * @param rdd
- * @param fname
- * @param outputFormat
- * @throws DMLRuntimeException
- */
- private void writeMatrixFromRDDtoHDFS(RDDObject rdd, String fname, String outputFormat)
- throws DMLRuntimeException
- {
- //prepare output info
- MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
- OutputInfo oinfo = (outputFormat != null ? OutputInfo.stringToOutputInfo (outputFormat)
- : InputInfo.getMatchingOutputInfo (iimd.getInputInfo ()));
-
- //note: the write of an RDD to HDFS might trigger
- //lazy evaluation of pending transformations.
- long newnnz = SparkExecutionContext.writeRDDtoHDFS(rdd, fname, oinfo);
- ((MatrixDimensionsMetaData) _metaData).getMatrixCharacteristics().setNonZeros(newnnz);
- }
-
- /**
- *
- * @param filePathAndName
- * @param rlen
- * @param clen
- * @return
- * @throws IOException
- */
- private MatrixBlock readMatrixFromHDFS(String filePathAndName, long rlen, long clen)
- throws IOException
- {
- long begin = 0;
-
- MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
- MatrixCharacteristics mc = iimd.getMatrixCharacteristics();
-
- if( LOG.isTraceEnabled() ){
- LOG.trace("Reading matrix from HDFS... " + getVarName() + " Path: " + filePathAndName
- + ", dimensions: [" + mc.getRows() + ", " + mc.getCols() + ", " + mc.getNonZeros() + "]");
- begin = System.currentTimeMillis();
- }
-
- double sparsity = ( mc.getNonZeros() >= 0 ? ((double)mc.getNonZeros())/(mc.getRows()*mc.getCols()) : 1.0d) ; //expected sparsity
- MatrixBlock newData = DataConverter.readMatrixFromHDFS(filePathAndName, iimd.getInputInfo(),
- rlen, clen, mc.getRowsPerBlock(), mc.getColsPerBlock(), sparsity, getFileFormatProperties());
-
- //sanity check correct output
- if( newData == null ) {
- throw new IOException("Unable to load matrix from file: "+filePathAndName);
- }
-
- if( LOG.isTraceEnabled() )
- LOG.trace("Reading Completed: " + (System.currentTimeMillis()-begin) + " msec.");
-
- return newData;
- }
-
- /**
* Writes in-memory matrix to HDFS in a specified format.
*
* @throws DMLRuntimeException
* @throws IOException
*/
- private void writeMatrixToHDFS (String filePathAndName, String outputFormat, int replication, FileFormatProperties formatProperties)
- throws DMLRuntimeException, IOException
+ @Override
+ protected void writeBlobToHDFS(String fname, String ofmt, int rep, FileFormatProperties fprop)
+ throws IOException, DMLRuntimeException
{
long begin = 0;
if( LOG.isTraceEnabled() ){
- LOG.trace (" Writing matrix to HDFS... " + getVarName() + " Path: " + filePathAndName + ", Format: " +
- (outputFormat != null ? outputFormat : "inferred from metadata"));
+ LOG.trace (" Writing matrix to HDFS... " + getVarName() + " Path: " + fname + ", Format: " +
+ (ofmt != null ? ofmt : "inferred from metadata"));
begin = System.currentTimeMillis();
}
@@ -1214,131 +661,43 @@ public class MatrixObject extends CacheableData<MatrixBlock>
// Get the dimension information from the metadata stored within MatrixObject
MatrixCharacteristics mc = iimd.getMatrixCharacteristics ();
// Write the matrix to HDFS in requested format
- OutputInfo oinfo = (outputFormat != null ? OutputInfo.stringToOutputInfo (outputFormat)
- : InputInfo.getMatchingOutputInfo (iimd.getInputInfo ()));
+ OutputInfo oinfo = (ofmt != null ? OutputInfo.stringToOutputInfo (ofmt) :
+ InputInfo.getMatchingOutputInfo (iimd.getInputInfo ()));
// when outputFormat is binaryblock, make sure that matrixCharacteristics has correct blocking dimensions
// note: this is only required if singlenode (due to binarycell default)
if ( oinfo == OutputInfo.BinaryBlockOutputInfo && DMLScript.rtplatform == RUNTIME_PLATFORM.SINGLE_NODE &&
(mc.getRowsPerBlock() != ConfigurationManager.getBlocksize() || mc.getColsPerBlock() != ConfigurationManager.getBlocksize()) )
{
- DataConverter.writeMatrixToHDFS(_data, filePathAndName, oinfo, new MatrixCharacteristics(mc.getRows(), mc.getCols(), ConfigurationManager.getBlocksize(), ConfigurationManager.getBlocksize(), mc.getNonZeros()), replication, formatProperties);
+ DataConverter.writeMatrixToHDFS(_data, fname, oinfo, new MatrixCharacteristics(mc.getRows(), mc.getCols(), ConfigurationManager.getBlocksize(), ConfigurationManager.getBlocksize(), mc.getNonZeros()), rep, fprop);
}
else {
- DataConverter.writeMatrixToHDFS(_data, filePathAndName, oinfo, mc, replication, formatProperties);
+ DataConverter.writeMatrixToHDFS(_data, fname, oinfo, mc, rep, fprop);
}
if( LOG.isTraceEnabled() )
- LOG.trace("Writing matrix to HDFS ("+filePathAndName+") - COMPLETED... " + (System.currentTimeMillis()-begin) + " msec.");
+ LOG.trace("Writing matrix to HDFS ("+fname+") - COMPLETED... " + (System.currentTimeMillis()-begin) + " msec.");
}
- else if( LOG.isTraceEnabled() )
- {
- LOG.trace ("Writing matrix to HDFS ("+filePathAndName+") - NOTHING TO WRITE (_data == null).");
+ else if( LOG.isTraceEnabled() ) {
+ LOG.trace ("Writing matrix to HDFS ("+fname+") - NOTHING TO WRITE (_data == null).");
}
if( DMLScript.STATISTICS )
CacheStatistics.incrementHDFSWrites();
}
- /**
- *
- * @param filePathAndName
- * @param outputFormat
- * @throws DMLRuntimeException
- * @throws IOException
- */
- private void writeMetaData (String filePathAndName, String outputFormat, FileFormatProperties formatProperties)
- throws DMLRuntimeException, IOException
- {
- MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
-
- if (iimd != null)
- {
- // Write the matrix to HDFS in requested format
- OutputInfo oinfo = (outputFormat != null ? OutputInfo.stringToOutputInfo (outputFormat)
- : InputInfo.getMatchingOutputInfo (iimd.getInputInfo ()));
-
- if ( oinfo != OutputInfo.MatrixMarketOutputInfo ) {
- // Get the dimension information from the metadata stored within MatrixObject
- MatrixCharacteristics mc = iimd.getMatrixCharacteristics ();
-
- // when outputFormat is binaryblock, make sure that matrixCharacteristics has correct blocking dimensions
- // note: this is only required if singlenode (due to binarycell default)
- if ( oinfo == OutputInfo.BinaryBlockOutputInfo && DMLScript.rtplatform == RUNTIME_PLATFORM.SINGLE_NODE &&
- (mc.getRowsPerBlock() != ConfigurationManager.getBlocksize() || mc.getColsPerBlock() != ConfigurationManager.getBlocksize()) )
- {
- mc = new MatrixCharacteristics(mc.getRows(), mc.getCols(), ConfigurationManager.getBlocksize(), ConfigurationManager.getBlocksize(), mc.getNonZeros());
- }
- MapReduceTool.writeMetaDataFile (filePathAndName + ".mtd", valueType, mc, oinfo, formatProperties);
- }
- }
- else {
- throw new DMLRuntimeException("Unexpected error while writing mtd file (" + filePathAndName + ") -- metadata is null.");
- }
- }
-
- /**
- *
- * @param outputFormat
- * @return
- */
- private boolean isEqualOutputFormat( String outputFormat )
+ @Override
+ protected void writeBlobFromRDDtoHDFS(RDDObject rdd, String fname, String outputFormat)
+ throws IOException, DMLRuntimeException
{
- boolean ret = true;
-
- if( outputFormat != null )
- {
- try
- {
- MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
- OutputInfo oi1 = InputInfo.getMatchingOutputInfo( iimd.getInputInfo() );
- OutputInfo oi2 = OutputInfo.stringToOutputInfo( outputFormat );
- if( oi1 != oi2 )
- {
- ret = false;
- }
- }
- catch(Exception ex)
- {
- ret = false;
- }
- }
-
- return ret;
- }
-
- /**
- *
- * @param add
- */
- private void updateStatusPinned(boolean add) {
- if( _data != null ) { //data should never be null
- long size = sizePinned.get();
- size += (add ? 1 : -1) * _data.getInMemorySize();
- sizePinned.set( Math.max(size,0) );
- }
- }
-
- /**
- *
- * @param flag
- */
- public void enableUpdateInPlace(boolean flag) {
- _updateInPlaceFlag = flag;
- }
-
- /**
- *
- * @return
- */
- public boolean isUpdateInPlaceEnabled() {
- return _updateInPlaceFlag;
- }
-
- /**
- *
- */
- public void setEmptyStatus() {
- setEmpty();
+ //prepare output info
+ MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
+ OutputInfo oinfo = (outputFormat != null ? OutputInfo.stringToOutputInfo (outputFormat)
+ : InputInfo.getMatchingOutputInfo (iimd.getInputInfo ()));
+
+ //note: the write of an RDD to HDFS might trigger
+ //lazy evaluation of pending transformations.
+ long newnnz = SparkExecutionContext.writeRDDtoHDFS(rdd, fname, oinfo);
+ ((MatrixDimensionsMetaData) _metaData).getMatrixCharacteristics().setNonZeros(newnnz);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/src/main/java/org/apache/sysml/runtime/instructions/spark/data/LineageObject.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/spark/data/LineageObject.java b/src/main/java/org/apache/sysml/runtime/instructions/spark/data/LineageObject.java
index bcf37bb..4b550cc 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/spark/data/LineageObject.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/spark/data/LineageObject.java
@@ -22,21 +22,19 @@ package org.apache.sysml.runtime.instructions.spark.data;
import java.util.ArrayList;
import java.util.List;
-import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
+import org.apache.sysml.runtime.controlprogram.caching.CacheableData;
public abstract class LineageObject
{
-
//basic lineage information
protected int _numRef = -1;
protected List<LineageObject> _childs = null;
protected String _varName = null;
- //N:1 back reference to matrix object
- protected MatrixObject _mo = null;
+ //N:1 back reference to matrix/frame object
+ protected CacheableData<?> _cd = null;
- protected LineageObject()
- {
+ protected LineageObject() {
_numRef = 0;
_childs = new ArrayList<LineageObject>();
}
@@ -45,38 +43,31 @@ public abstract class LineageObject
return _varName;
}
- public int getNumReferences()
- {
+ public int getNumReferences() {
return _numRef;
}
- public void setBackReference(MatrixObject mo)
- {
- _mo = mo;
+ public void setBackReference(CacheableData<?> cd) {
+ _cd = cd;
}
- public boolean hasBackReference()
- {
- return (_mo != null);
+ public boolean hasBackReference() {
+ return (_cd != null);
}
- public void incrementNumReferences()
- {
+ public void incrementNumReferences() {
_numRef++;
}
- public void decrementNumReferences()
- {
+ public void decrementNumReferences() {
_numRef--;
}
- public List<LineageObject> getLineageChilds()
- {
+ public List<LineageObject> getLineageChilds() {
return _childs;
}
- public void addLineageChild(LineageObject lob)
- {
+ public void addLineageChild(LineageObject lob) {
lob.incrementNumReferences();
_childs.add( lob );
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/src/main/java/org/apache/sysml/runtime/matrix/data/FrameBlock.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/data/FrameBlock.java b/src/main/java/org/apache/sysml/runtime/matrix/data/FrameBlock.java
index 7fca641..0e86cbe 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/data/FrameBlock.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/data/FrameBlock.java
@@ -407,6 +407,11 @@ public class FrameBlock implements Writable, CacheBlock, Externalizable
return true;
}
+ @Override
+ public void compactEmptyBlock() {
+ //do nothing
+ }
+
///////
// indexing and append operations
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/1fd254ff/src/main/java/org/apache/sysml/runtime/matrix/data/MatrixBlock.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/data/MatrixBlock.java b/src/main/java/org/apache/sysml/runtime/matrix/data/MatrixBlock.java
index 5ce255c..f62fbc6 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/data/MatrixBlock.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/data/MatrixBlock.java
@@ -2781,6 +2781,12 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
|| (sparse && sparseBlock instanceof SparseBlockCSR);
}
+ @Override
+ public void compactEmptyBlock() {
+ if( isEmptyBlock(false) && isAllocated() )
+ cleanupBlock(true, true);
+ }
+
////////
// Core block operations (called from instructions)
[2/2] incubator-systemml git commit: [SYSTEMML-561] Integrated CP
operations over frames (compiler/runtime)
Posted by mb...@apache.org.
[SYSTEMML-561] Integrated CP operations over frames (compiler/runtime)
So far transformencode, transformapply, transformdecode as well as other
operations over frames were only usable through JMLC due to the missing
full compiler/runtime integration of frames. This patch overcomes this
limitation and makes these operations available in CP. Until full
support for distributed operations, we always compile CP operations
though. Furthermore, this change also (1) consolidates the input
variable handling of the old transform by datatype-specific createvar
handling, and (2) fixes hop deep copy issues with multi-assign builtin
functions (e.g., qr, lu, eigen).
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/7528146c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/7528146c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/7528146c
Branch: refs/heads/master
Commit: 7528146ceee4c62255ee64e5cd8c8359d7593dc2
Parents: 1fd254f
Author: Matthias Boehm <mb...@us.ibm.com>
Authored: Tue Apr 12 23:39:11 2016 -0700
Committer: Matthias Boehm <mb...@us.ibm.com>
Committed: Wed Apr 13 11:34:46 2016 -0700
----------------------------------------------------------------------
src/main/java/org/apache/sysml/hops/DataOp.java | 4 +-
.../java/org/apache/sysml/hops/FunctionOp.java | 10 +-
.../sysml/hops/ParameterizedBuiltinOp.java | 3 +-
.../apache/sysml/hops/cost/CostEstimator.java | 10 +-
src/main/java/org/apache/sysml/lops/Data.java | 13 +-
src/main/java/org/apache/sysml/lops/Lop.java | 1 +
.../java/org/apache/sysml/lops/compile/Dag.java | 19 +--
.../org/apache/sysml/parser/DataExpression.java | 5 -
.../ExternalFunctionProgramBlock.java | 21 +--
.../controlprogram/caching/CacheableData.java | 10 +-
.../controlprogram/caching/FrameObject.java | 48 +++++-
.../controlprogram/caching/MatrixObject.java | 34 +---
.../context/ExecutionContext.java | 40 ++++-
.../context/SparkExecutionContext.java | 95 ++++++++++-
.../runtime/instructions/MRJobInstruction.java | 9 +
.../cp/ParameterizedBuiltinCPInstruction.java | 5 +-
.../instructions/cp/VariableCPInstruction.java | 113 ++++++------
.../ParameterizedBuiltinSPInstruction.java | 3 +-
.../sysml/runtime/io/FrameReaderFactory.java | 4 -
.../sysml/runtime/transform/DataTransform.java | 89 +++++-----
.../transform/TransformEncodeDecodeTest.java | 170 +++++++++++++++++++
.../transform/TransformEncodeDecode.dml | 33 ++++
.../transform/TransformEncodeDecodeSpec.json | 5 +
.../functions/frame/ZPackageSuite.java | 1 +
.../functions/jmlc/ZPackageSuite.java | 4 +
.../functions/transform/ZPackageSuite.java | 1 +
26 files changed, 554 insertions(+), 196 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/hops/DataOp.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/DataOp.java b/src/main/java/org/apache/sysml/hops/DataOp.java
index bdaa3b9..c907e33 100644
--- a/src/main/java/org/apache/sysml/hops/DataOp.java
+++ b/src/main/java/org/apache/sysml/hops/DataOp.java
@@ -435,8 +435,8 @@ public class DataOp extends Hop
{
checkAndSetForcedPlatform();
- //additional check for write only
- if( getDataType()==DataType.SCALAR )
+ //additional check for write only (TODO: remove frame here once support for distributed)
+ if( getDataType()==DataType.SCALAR || getDataType()==DataType.FRAME )
_etypeForced = ExecType.CP;
if( _etypeForced != null )
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/hops/FunctionOp.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/FunctionOp.java b/src/main/java/org/apache/sysml/hops/FunctionOp.java
index 9da8084..ae199b5 100644
--- a/src/main/java/org/apache/sysml/hops/FunctionOp.java
+++ b/src/main/java/org/apache/sysml/hops/FunctionOp.java
@@ -136,7 +136,6 @@ public class FunctionOp extends Hop
_processingMemEstimate = computeIntermediateMemEstimate(_dim1, _dim2, lnnz);
}
_memEstimate = getInputOutputSize();
- //System.out.println("QREst " + (_memEstimate/1024/1024));
}
}
@@ -150,25 +149,19 @@ public class FunctionOp extends Hop
// upper-triangular and lower-triangular matrices
long outputH = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(0).getDim1(), getOutputs().get(0).getDim2(), 0.5);
long outputR = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(1).getDim1(), getOutputs().get(1).getDim2(), 0.5);
- //System.out.println("QROut " + (outputH+outputR)/1024/1024);
return outputH+outputR;
-
}
else if ( getFunctionName().equalsIgnoreCase("lu") ) {
// upper-triangular and lower-triangular matrices
long outputP = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(1).getDim1(), getOutputs().get(1).getDim2(), 1.0/getOutputs().get(1).getDim2());
long outputL = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(0).getDim1(), getOutputs().get(0).getDim2(), 0.5);
long outputU = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(1).getDim1(), getOutputs().get(1).getDim2(), 0.5);
- //System.out.println("LUOut " + (outputL+outputU+outputP)/1024/1024);
return outputL+outputU+outputP;
-
}
else if ( getFunctionName().equalsIgnoreCase("eigen") ) {
long outputVectors = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(0).getDim1(), getOutputs().get(0).getDim2(), 1.0);
long outputValues = OptimizerUtils.estimateSizeExactSparsity(getOutputs().get(1).getDim1(), 1, 1.0);
- //System.out.println("EigenOut " + (outputVectors+outputValues)/1024/1024);
return outputVectors+outputValues;
-
}
else
throw new RuntimeException("Invalid call of computeOutputMemEstimate in FunctionOp.");
@@ -275,6 +268,7 @@ public class FunctionOp extends Hop
}
@Override
+ @SuppressWarnings("unchecked")
public Object clone() throws CloneNotSupportedException
{
FunctionOp ret = new FunctionOp();
@@ -287,6 +281,8 @@ public class FunctionOp extends Hop
ret._fnamespace = _fnamespace;
ret._fname = _fname;
ret._outputs = _outputs.clone();
+ if( _outputHops != null )
+ ret._outputHops = (ArrayList<Hop>) _outputHops.clone();
return ret;
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/hops/ParameterizedBuiltinOp.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/ParameterizedBuiltinOp.java b/src/main/java/org/apache/sysml/hops/ParameterizedBuiltinOp.java
index c94d51c..72f28f8 100644
--- a/src/main/java/org/apache/sysml/hops/ParameterizedBuiltinOp.java
+++ b/src/main/java/org/apache/sysml/hops/ParameterizedBuiltinOp.java
@@ -996,7 +996,8 @@ public class ParameterizedBuiltinOp extends Hop implements MultiThreadedHop
// force remote, at runtime cp transform triggered for small files.
return REMOTE;
}
- else if( _op == ParamBuiltinOp.TRANSFORMAPPLY ) {
+ else if( _op == ParamBuiltinOp.TRANSFORMAPPLY
+ || _op == ParamBuiltinOp.TRANSFORMDECODE ) {
return ExecType.CP;
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/hops/cost/CostEstimator.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/hops/cost/CostEstimator.java b/src/main/java/org/apache/sysml/hops/cost/CostEstimator.java
index 6b9c21f..7e46d71 100644
--- a/src/main/java/org/apache/sysml/hops/cost/CostEstimator.java
+++ b/src/main/java/org/apache/sysml/hops/cost/CostEstimator.java
@@ -324,11 +324,11 @@ public abstract class CostEstimator
if( parts.length<10 )
return;
String varname = parts[1];
- long rlen = Long.parseLong(parts[5]);
- long clen = Long.parseLong(parts[6]);
- long brlen = Long.parseLong(parts[7]);
- long bclen = Long.parseLong(parts[8]);
- long nnz = Long.parseLong(parts[9]);
+ long rlen = Long.parseLong(parts[6]);
+ long clen = Long.parseLong(parts[7]);
+ long brlen = Long.parseLong(parts[8]);
+ long bclen = Long.parseLong(parts[9]);
+ long nnz = Long.parseLong(parts[10]);
VarStats vs = new VarStats(rlen, clen, brlen, bclen, nnz, false);
stats.put(varname, vs);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/lops/Data.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/lops/Data.java b/src/main/java/org/apache/sysml/lops/Data.java
index 24b2a03..b464efa 100644
--- a/src/main/java/org/apache/sysml/lops/Data.java
+++ b/src/main/java/org/apache/sysml/lops/Data.java
@@ -432,7 +432,7 @@ public class Data extends Lop
if ( operation == OperationTypes.WRITE ) {
sb.append( OPERAND_DELIMITOR );
String fmt = "";
- if ( getDataType() == DataType.MATRIX ) {
+ if ( getDataType() == DataType.MATRIX || getDataType() == DataType.FRAME ) {
if ( oparams.getFormat() == Format.MM )
fmt = "matrixmarket";
else if (oparams.getFormat() == Format.TEXT)
@@ -522,18 +522,15 @@ public class Data extends Lop
OutputParameters oparams = getOutputParameters();
String fmt = "";
- // TODO: following logic should change once we LOPs encode key-value-class information.
if ( oparams.getFormat() == Format.TEXT )
fmt = "textcell";
else if ( oparams.getFormat() == Format.MM )
fmt = "matrixmarket";
else if ( oparams.getFormat() == Format.CSV )
fmt = "csv";
- else {
- if ( oparams.getRowsInBlock() > 0 || oparams.getColsInBlock() > 0 )
- fmt = "binaryblock";
- else
- fmt = "binarycell";
+ else { //binary
+ fmt = ( getDataType() == DataType.FRAME || oparams.getRowsInBlock() > 0
+ || oparams.getColsInBlock() > 0 ) ? "binaryblock" : "binarycell";
}
StringBuilder sb = new StringBuilder();
@@ -546,6 +543,8 @@ public class Data extends Lop
sb.append( outputFileName );
sb.append( OPERAND_DELIMITOR );
sb.append( false );
+ sb.append( OPERAND_DELIMITOR );
+ sb.append( getDataType() );
sb.append( OPERAND_DELIMITOR ); // only persistent reads come here!
sb.append( fmt );
sb.append( OPERAND_DELIMITOR );
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/lops/Lop.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/lops/Lop.java b/src/main/java/org/apache/sysml/lops/Lop.java
index 1c2bd5b..59d11be 100644
--- a/src/main/java/org/apache/sysml/lops/Lop.java
+++ b/src/main/java/org/apache/sysml/lops/Lop.java
@@ -87,6 +87,7 @@ public abstract class Lop
public static final String NAME_VALUE_SEPARATOR = "="; // e.g., used in parameterized builtins
public static final String MATRIX_VAR_NAME_PREFIX = "_mVar";
+ public static final String FRAME_VAR_NAME_PREFIX = "_fVar";
public static final String SCALAR_VAR_NAME_PREFIX = "_Var";
// Boolean array to hold the list of nodes(lops) in the DAG that are reachable from this lop.
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/lops/compile/Dag.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/lops/compile/Dag.java b/src/main/java/org/apache/sysml/lops/compile/Dag.java
index dba852b..6ec3a5a 100644
--- a/src/main/java/org/apache/sysml/lops/compile/Dag.java
+++ b/src/main/java/org/apache/sysml/lops/compile/Dag.java
@@ -2357,10 +2357,8 @@ public class Dag<N extends Lop>
DataExpression.DELIM_DELIMITER, DataExpression.DEFAULT_DELIM_DELIMITER);
Instruction createvarInst = VariableCPInstruction.prepareCreateVariableInstruction(
- oparams.getLabel(),
- oparams.getFile_name(),
- true,
- OutputInfo.outputInfoToString(OutputInfo.CSVOutputInfo),
+ oparams.getLabel(), oparams.getFile_name(), true,
+ DataType.MATRIX, OutputInfo.outputInfoToString(OutputInfo.CSVOutputInfo),
new MatrixCharacteristics(oparams.getNumRows(), oparams.getNumCols(), -1, -1, oparams.getNnz()), oparams.getUpdateInPlace(),
false, delimLop.getStringValue(), true
);
@@ -2389,7 +2387,8 @@ public class Dag<N extends Lop>
// generate temporary filename and a variable name to hold the
// output produced by "rootNode"
oparams.setFile_name(getFilePath() + "temp" + job_id.getNextID());
- oparams.setLabel(Lop.MATRIX_VAR_NAME_PREFIX + var_index.getNextID());
+ oparams.setLabel( (node.getDataType()==DataType.MATRIX ? Lop.MATRIX_VAR_NAME_PREFIX :
+ Lop.FRAME_VAR_NAME_PREFIX) + var_index.getNextID());
// generate an instruction that creates a symbol table entry for the new variable
//String createInst = prepareVariableInstruction("createvar", node);
@@ -2399,7 +2398,7 @@ public class Dag<N extends Lop>
Instruction createvarInst = VariableCPInstruction.prepareCreateVariableInstruction(
oparams.getLabel(),
oparams.getFile_name(),
- true,
+ true, node.getDataType(),
OutputInfo.outputInfoToString(getOutputInfo(node, false)),
new MatrixCharacteristics(oparams.getNumRows(), oparams.getNumCols(), rpb, cpb, oparams.getNnz()),
oparams.getUpdateInPlace()
@@ -2431,7 +2430,7 @@ public class Dag<N extends Lop>
Instruction createvarInst = VariableCPInstruction.prepareCreateVariableInstruction(
fnOutParams.getLabel(),
getFilePath() + fnOutParams.getLabel(),
- true,
+ true, fnOut.getDataType(),
OutputInfo.outputInfoToString(getOutputInfo(fnOut, false)),
new MatrixCharacteristics(fnOutParams.getNumRows(), fnOutParams.getNumCols(), (int)fnOutParams.getRowsInBlock(), (int)fnOutParams.getColsInBlock(), fnOutParams.getNnz()),
oparams.getUpdateInPlace()
@@ -2547,7 +2546,7 @@ public class Dag<N extends Lop>
Instruction createvarInst = VariableCPInstruction.prepareCreateVariableInstruction(
tempVarName,
tempFileName,
- true,
+ true, node.getDataType(),
OutputInfo.outputInfoToString(out.getOutInfo()),
new MatrixCharacteristics(oparams.getNumRows(), oparams.getNumCols(), rpb, cpb, oparams.getNnz()),
oparams.getUpdateInPlace()
@@ -2657,7 +2656,7 @@ public class Dag<N extends Lop>
createvarInst= VariableCPInstruction.prepareCreateVariableInstruction(
oparams.getLabel(),
tempFileName,
- false,
+ false, node.getDataType(),
OutputInfo.outputInfoToString(getOutputInfo(node, false)),
new MatrixCharacteristics(oparams.getNumRows(), oparams.getNumCols(), rpb, cpb, oparams.getNnz()),
oparams.getUpdateInPlace()
@@ -2685,7 +2684,7 @@ public class Dag<N extends Lop>
createvarInst= VariableCPInstruction.prepareCreateVariableInstruction(
oparams.getLabel(),
fnameStr,
- false,
+ false, node.getDataType(),
OutputInfo.outputInfoToString(getOutputInfo(node, false)),
new MatrixCharacteristics(oparams.getNumRows(), oparams.getNumCols(), rpb, cpb, oparams.getNnz()),
oparams.getUpdateInPlace()
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/parser/DataExpression.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/parser/DataExpression.java b/src/main/java/org/apache/sysml/parser/DataExpression.java
index 562e2ce..e1576c6 100644
--- a/src/main/java/org/apache/sysml/parser/DataExpression.java
+++ b/src/main/java/org/apache/sysml/parser/DataExpression.java
@@ -914,11 +914,6 @@ public class DataExpression extends DataIdentifier
}
}
- // Table must be in CSV format
- if ( !isMatrix && !getVarParam(FORMAT_TYPE).toString().equalsIgnoreCase("csv") ) {
- raiseValidateError("Input data of type 'table' must be in CSV format. Invalid format '" + getVarParam(FORMAT_TYPE)+ "' in statement: " + this.toString(), conditional);
- }
-
// initialize block dimensions to UNKNOWN
getOutput().setBlockDimensions(-1, -1);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/controlprogram/ExternalFunctionProgramBlock.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/ExternalFunctionProgramBlock.java b/src/main/java/org/apache/sysml/runtime/controlprogram/ExternalFunctionProgramBlock.java
index 1485e51..d01b444 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/ExternalFunctionProgramBlock.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/ExternalFunctionProgramBlock.java
@@ -392,15 +392,7 @@ public class ExternalFunctionProgramBlock extends FunctionProgramBlock
}
// create metadata instructions to populate symbol table
// with variables that hold blocked matrices
-
- /*StringBuilder mtdInst = new StringBuilder();
- mtdInst.append("CP" + Lops.OPERAND_DELIMITOR + "createvar");
- mtdInst.append(Lops.OPERAND_DELIMITOR + outLabels.get(i) + Lops.DATATYPE_PREFIX + matrices.get(i).getDataType() + Lops.VALUETYPE_PREFIX + matrices.get(i).getValueType());
- mtdInst.append(Lops.OPERAND_DELIMITOR + outputs[i] + Lops.DATATYPE_PREFIX + DataType.SCALAR + Lops.VALUETYPE_PREFIX + ValueType.STRING);
- mtdInst.append(Lops.OPERAND_DELIMITOR + OutputInfo.outputInfoToString(OutputInfo.BinaryBlockOutputInfo) ) ;
- c2binst.add(CPInstructionParser.parseSingleInstruction(mtdInst.toString()));*/
- Instruction createInst = VariableCPInstruction.prepareCreateVariableInstruction(outLabels.get(i), outputs[i], false, OutputInfo.outputInfoToString(OutputInfo.BinaryBlockOutputInfo));
-
+ Instruction createInst = VariableCPInstruction.prepareCreateMatrixVariableInstruction(outLabels.get(i), outputs[i], false, OutputInfo.outputInfoToString(OutputInfo.BinaryBlockOutputInfo));
createInst.setLocation(matrices.get(i));
c2binst.add(createInst);
@@ -514,16 +506,7 @@ public class ExternalFunctionProgramBlock extends FunctionProgramBlock
}
// create metadata instructions to populate symbol table
// with variables that hold unblocked matrices
-
- /*StringBuilder mtdInst = new StringBuilder();
- mtdInst.append("CP" + Lops.OPERAND_DELIMITOR + "createvar");
- mtdInst.append(Lops.OPERAND_DELIMITOR + outLabels.get(i) + Lops.DATATYPE_PREFIX + matrices.get(i).getDataType() + Lops.VALUETYPE_PREFIX + matrices.get(i).getValueType());
- mtdInst.append(Lops.OPERAND_DELIMITOR + outputs[i] + Lops.DATATYPE_PREFIX + DataType.SCALAR + Lops.VALUETYPE_PREFIX + ValueType.STRING);
- mtdInst.append(Lops.OPERAND_DELIMITOR + OutputInfo.outputInfoToString(OutputInfo.TextCellOutputInfo) ) ;
- b2cinst.add(CPInstructionParser.parseSingleInstruction(mtdInst.toString()));*/
-
- Instruction createInst = VariableCPInstruction.prepareCreateVariableInstruction(outLabels.get(i), outputs[i], false, OutputInfo.outputInfoToString(OutputInfo.TextCellOutputInfo));
-
+ Instruction createInst = VariableCPInstruction.prepareCreateMatrixVariableInstruction(outLabels.get(i), outputs[i], false, OutputInfo.outputInfoToString(OutputInfo.TextCellOutputInfo));
createInst.setLocation(matrices.get(i));
b2cinst.add(createInst);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
index c75820f..d2811a0 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/CacheableData.java
@@ -38,6 +38,7 @@ import org.apache.sysml.runtime.instructions.cp.Data;
import org.apache.sysml.runtime.instructions.spark.data.BroadcastObject;
import org.apache.sysml.runtime.instructions.spark.data.RDDObject;
import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
+import org.apache.sysml.runtime.matrix.MatrixDimensionsMetaData;
import org.apache.sysml.runtime.matrix.MatrixFormatMetaData;
import org.apache.sysml.runtime.matrix.MetaData;
import org.apache.sysml.runtime.matrix.data.FileFormatProperties;
@@ -324,6 +325,11 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
_metaData = null;
}
+ public MatrixCharacteristics getMatrixCharacteristics() {
+ MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
+ return meta.getMatrixCharacteristics();
+ }
+
/**
*
*/
@@ -949,7 +955,9 @@ public abstract class CacheableData<T extends CacheBlock> extends Data
* @return
* @throws IOException
*/
- protected T readBlobFromHDFS(String fname) throws IOException {
+ protected T readBlobFromHDFS(String fname)
+ throws IOException
+ {
MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
MatrixCharacteristics mc = iimd.getMatrixCharacteristics();
return readBlobFromHDFS(fname, mc.getRows(), mc.getCols());
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
index f83e7c1..d884054 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/caching/FrameObject.java
@@ -27,11 +27,17 @@ import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.runtime.DMLRuntimeException;
import org.apache.sysml.runtime.instructions.spark.data.RDDObject;
+import org.apache.sysml.runtime.io.FrameReader;
+import org.apache.sysml.runtime.io.FrameReaderFactory;
+import org.apache.sysml.runtime.io.FrameWriter;
+import org.apache.sysml.runtime.io.FrameWriterFactory;
import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
import org.apache.sysml.runtime.matrix.MatrixDimensionsMetaData;
+import org.apache.sysml.runtime.matrix.MatrixFormatMetaData;
import org.apache.sysml.runtime.matrix.MetaData;
import org.apache.sysml.runtime.matrix.data.FileFormatProperties;
import org.apache.sysml.runtime.matrix.data.FrameBlock;
+import org.apache.sysml.runtime.matrix.data.OutputInfo;
public class FrameObject extends CacheableData<FrameBlock>
{
@@ -84,6 +90,23 @@ public class FrameObject extends CacheableData<FrameBlock>
mc.setDimension( _data.getNumRows(),_data.getNumColumns() );
}
+ /**
+ *
+ * @return
+ */
+ public long getNumRows() {
+ MatrixCharacteristics mc = getMatrixCharacteristics();
+ return mc.getRows();
+ }
+
+ /**
+ *
+ * @return
+ */
+ public long getNumColumns() {
+ MatrixCharacteristics mc = getMatrixCharacteristics();
+ return mc.getCols();
+ }
@Override
protected FrameBlock readBlobFromCache(String fname) throws IOException {
@@ -94,13 +117,31 @@ public class FrameObject extends CacheableData<FrameBlock>
protected FrameBlock readBlobFromHDFS(String fname, long rlen, long clen)
throws IOException
{
- throw new IOException("Not implemented yet.");
+ MatrixFormatMetaData iimd = (MatrixFormatMetaData) _metaData;
+ MatrixCharacteristics mc = iimd.getMatrixCharacteristics();
+
+ FrameBlock data = null;
+ try {
+ FrameReader reader = FrameReaderFactory.createFrameReader(
+ iimd.getInputInfo());
+ data = reader.readFrameFromHDFS(fname, mc.getRows(), mc.getCols());
+ }
+ catch( DMLRuntimeException ex ) {
+ throw new IOException(ex);
+ }
+
+ //sanity check correct output
+ if( data == null )
+ throw new IOException("Unable to load frame from file: "+fname);
+
+ return data;
}
@Override
protected FrameBlock readBlobFromRDD(RDDObject rdd, MutableBoolean status)
throws IOException
{
+ //TODO support for distributed frame representations
throw new IOException("Not implemented yet.");
}
@@ -108,13 +149,16 @@ public class FrameObject extends CacheableData<FrameBlock>
protected void writeBlobToHDFS(String fname, String ofmt, int rep, FileFormatProperties fprop)
throws IOException, DMLRuntimeException
{
- throw new IOException("Not implemented yet.");
+ OutputInfo oinfo = OutputInfo.stringToOutputInfo(ofmt);
+ FrameWriter writer = FrameWriterFactory.createFrameWriter(oinfo);
+ writer.writeFrameToHDFS(_data, fname, getNumRows(), getNumColumns());
}
@Override
protected void writeBlobFromRDDtoHDFS(RDDObject rdd, String fname, String ofmt)
throws IOException, DMLRuntimeException
{
+ //TODO support for distributed frame representations
throw new IOException("Not implemented yet.");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/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 cb27ee4..639653a 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
@@ -150,14 +150,13 @@ public class MatrixObject extends CacheableData<MatrixBlock>
_data.getNumColumns() );
mc.setNonZeros( _data.getNonZeros() );
}
-
+
/**
*
* @return
*/
public long getNumRows() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
+ MatrixCharacteristics mc = getMatrixCharacteristics();
return mc.getRows ();
}
@@ -166,18 +165,16 @@ public class MatrixObject extends CacheableData<MatrixBlock>
* @return
*/
public long getNumColumns() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
+ MatrixCharacteristics mc = getMatrixCharacteristics();
return mc.getCols ();
}
-
+
/**
*
* @return
*/
public long getNumRowsPerBlock() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
+ MatrixCharacteristics mc = getMatrixCharacteristics();
return mc.getRowsPerBlock();
}
@@ -186,8 +183,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
* @return
*/
public long getNumColumnsPerBlock() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
+ MatrixCharacteristics mc = getMatrixCharacteristics();
return mc.getColsPerBlock();
}
@@ -196,8 +192,7 @@ public class MatrixObject extends CacheableData<MatrixBlock>
* @return
*/
public long getNnz() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
+ MatrixCharacteristics mc = getMatrixCharacteristics();
return mc.getNonZeros();
}
@@ -206,20 +201,10 @@ public class MatrixObject extends CacheableData<MatrixBlock>
* @return
*/
public double getSparsity() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- MatrixCharacteristics mc = meta.getMatrixCharacteristics();
+ MatrixCharacteristics mc = getMatrixCharacteristics();
return ((double)mc.getNonZeros())/mc.getRows()/mc.getCols();
}
- /**
- *
- * @return
- */
- public MatrixCharacteristics getMatrixCharacteristics() {
- MatrixDimensionsMetaData meta = (MatrixDimensionsMetaData) _metaData;
- return meta.getMatrixCharacteristics();
- }
-
public String toString()
{
StringBuilder str = new StringBuilder();
@@ -558,9 +543,8 @@ public class MatrixObject extends CacheableData<MatrixBlock>
mc.getRowsPerBlock(), mc.getColsPerBlock(), sparsity, getFileFormatProperties());
//sanity check correct output
- if( newData == null ) {
+ if( newData == null )
throw new IOException("Unable to load matrix from file: "+fname);
- }
if( LOG.isTraceEnabled() )
LOG.trace("Reading Completed: " + (System.currentTimeMillis()-begin) + " msec.");
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/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 f56a870..f2b34c2 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
@@ -33,6 +33,7 @@ import org.apache.sysml.runtime.DMLRuntimeException;
import org.apache.sysml.runtime.controlprogram.LocalVariableMap;
import org.apache.sysml.runtime.controlprogram.Program;
import org.apache.sysml.runtime.controlprogram.caching.CacheException;
+import org.apache.sysml.runtime.controlprogram.caching.CacheableData;
import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.instructions.Instruction;
@@ -45,12 +46,9 @@ import org.apache.sysml.runtime.instructions.cp.ScalarObject;
import org.apache.sysml.runtime.instructions.cp.StringObject;
import org.apache.sysml.runtime.matrix.MatrixCharacteristics;
import org.apache.sysml.runtime.matrix.MatrixDimensionsMetaData;
-import org.apache.sysml.runtime.matrix.MatrixFormatMetaData;
import org.apache.sysml.runtime.matrix.MetaData;
import org.apache.sysml.runtime.matrix.data.FrameBlock;
-import org.apache.sysml.runtime.matrix.data.InputInfo;
import org.apache.sysml.runtime.matrix.data.MatrixBlock;
-import org.apache.sysml.runtime.matrix.data.OutputInfo;
import org.apache.sysml.runtime.util.MapReduceTool;
import org.apache.sysml.runtime.util.UtilFunctions;
@@ -166,6 +164,34 @@ public class ExecutionContext
return (MatrixObject) dat;
}
+ public FrameObject getFrameObject(String varname)
+ throws DMLRuntimeException
+ {
+ Data dat = getVariable(varname);
+
+ //error handling if non existing or no matrix
+ if( dat == null )
+ throw new DMLRuntimeException("Variable '"+varname+"' does not exist in the symbol table.");
+ if( !(dat instanceof FrameObject) )
+ throw new DMLRuntimeException("Variable '"+varname+"' is not a frame.");
+
+ return (FrameObject) dat;
+ }
+
+ public CacheableData<?> getCacheableData(String varname)
+ throws DMLRuntimeException
+ {
+ Data dat = getVariable(varname);
+
+ //error handling if non existing or no matrix
+ if( dat == null )
+ throw new DMLRuntimeException("Variable '"+varname+"' does not exist in the symbol table.");
+ if( !(dat instanceof CacheableData<?>) )
+ throw new DMLRuntimeException("Variable '"+varname+"' is not a matrix or frame.");
+
+ return (CacheableData<?>) dat;
+ }
+
public MatrixCharacteristics getMatrixCharacteristics( String varname )
throws DMLRuntimeException
{
@@ -311,13 +337,9 @@ public class ExecutionContext
public void setFrameOutput(String varName, FrameBlock outputData)
throws DMLRuntimeException
{
- //TODO: fix createvar instructions generation for frame handles
- //FrameObject fo = (FrameObject) getVariable(varName);
- MatrixCharacteristics mc = new MatrixCharacteristics(-1, -1, -1, -1);
- MatrixFormatMetaData meta = new MatrixFormatMetaData(mc, OutputInfo.BinaryCellOutputInfo, InputInfo.BinaryCellInputInfo);
- FrameObject fo = new FrameObject(varName, meta);
+ FrameObject fo = (FrameObject) getVariable(varName);
fo.acquireModify(outputData);
- fo.release();
+ fo.release();
setVariable(varName, fo);
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/controlprogram/context/SparkExecutionContext.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/context/SparkExecutionContext.java b/src/main/java/org/apache/sysml/runtime/controlprogram/context/SparkExecutionContext.java
index 66bee85..d3134b3 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/context/SparkExecutionContext.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/context/SparkExecutionContext.java
@@ -43,6 +43,7 @@ import org.apache.sysml.hops.OptimizerUtils;
import org.apache.sysml.lops.Checkpoint;
import org.apache.sysml.runtime.DMLRuntimeException;
import org.apache.sysml.runtime.controlprogram.Program;
+import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
import org.apache.sysml.runtime.instructions.spark.CheckpointSPInstruction;
@@ -336,6 +337,92 @@ public class SparkExecutionContext extends ExecutionContext
}
/**
+ * FIXME: currently this implementation assumes matrix representations but frame signature
+ * in order to support the old transform implementation.
+ *
+ * @param mo
+ * @param inputInfo
+ * @return
+ * @throws DMLRuntimeException
+ */
+ @SuppressWarnings("unchecked")
+ public JavaPairRDD<?,?> getRDDHandleForFrameObject( FrameObject fo, InputInfo inputInfo )
+ throws DMLRuntimeException
+ {
+ //NOTE: MB this logic should be integrated into MatrixObject
+ //However, for now we cannot assume that spark libraries are
+ //always available and hence only store generic references in
+ //matrix object while all the logic is in the SparkExecContext
+
+ JavaPairRDD<?,?> rdd = null;
+ //CASE 1: rdd already existing (reuse if checkpoint or trigger
+ //pending rdd operations if not yet cached but prevent to re-evaluate
+ //rdd operations if already executed and cached
+ if( fo.getRDDHandle()!=null
+ && (fo.getRDDHandle().isCheckpointRDD() || !fo.isCached(false)) )
+ {
+ //return existing rdd handling (w/o input format change)
+ rdd = fo.getRDDHandle().getRDD();
+ }
+ //CASE 2: dirty in memory data or cached result of rdd operations
+ else if( fo.isDirty() || fo.isCached(false) )
+ {
+ //get in-memory matrix block and parallelize it
+ //w/ guarded parallelize (fallback to export, rdd from file if too large)
+ boolean fromFile = false;
+ if( !OptimizerUtils.checkSparkCollectMemoryBudget(fo.getMatrixCharacteristics(), 0) ) {
+ if( fo.isDirty() ) { //write only if necessary
+ fo.exportData();
+ }
+ rdd = getSparkContext().hadoopFile( fo.getFileName(), inputInfo.inputFormatClass, inputInfo.inputKeyClass, inputInfo.inputValueClass);
+ rdd = ((JavaPairRDD<MatrixIndexes, MatrixBlock>)rdd).mapToPair( new CopyBlockPairFunction() ); //cp is workaround for read bug
+ fromFile = true;
+ }
+ else { //default case
+ //MatrixBlock mb = mo.acquireRead(); //pin matrix in memory
+ //rdd = toJavaPairRDD(getSparkContext(), mb, (int)mo.getNumRowsPerBlock(), (int)mo.getNumColumnsPerBlock());
+ //fo.release(); //unpin matrix
+ throw new RuntimeException("Not implemented yet.");
+ }
+
+ //keep rdd handle for future operations on it
+ RDDObject rddhandle = new RDDObject(rdd, fo.getVarName());
+ rddhandle.setHDFSFile(fromFile);
+ fo.setRDDHandle(rddhandle);
+ }
+ //CASE 3: non-dirty (file exists on HDFS)
+ else
+ {
+ // parallelize hdfs-resident file
+ // For binary block, these are: SequenceFileInputFormat.class, MatrixIndexes.class, MatrixBlock.class
+ if(inputInfo == InputInfo.BinaryBlockInputInfo) {
+ rdd = getSparkContext().hadoopFile( fo.getFileName(), inputInfo.inputFormatClass, inputInfo.inputKeyClass, inputInfo.inputValueClass);
+ //note: this copy is still required in Spark 1.4 because spark hands out whatever the inputformat
+ //recordreader returns; the javadoc explicitly recommend to copy all key/value pairs
+ rdd = ((JavaPairRDD<MatrixIndexes, MatrixBlock>)rdd).mapToPair( new CopyBlockPairFunction() ); //cp is workaround for read bug
+ }
+ else if(inputInfo == InputInfo.TextCellInputInfo || inputInfo == InputInfo.CSVInputInfo || inputInfo == InputInfo.MatrixMarketInputInfo) {
+ rdd = getSparkContext().hadoopFile( fo.getFileName(), inputInfo.inputFormatClass, inputInfo.inputKeyClass, inputInfo.inputValueClass);
+ rdd = ((JavaPairRDD<LongWritable, Text>)rdd).mapToPair( new CopyTextInputFunction() ); //cp is workaround for read bug
+ }
+ else if(inputInfo == InputInfo.BinaryCellInputInfo) {
+ rdd = getSparkContext().hadoopFile( fo.getFileName(), inputInfo.inputFormatClass, inputInfo.inputKeyClass, inputInfo.inputValueClass);
+ rdd = ((JavaPairRDD<MatrixIndexes, MatrixCell>)rdd).mapToPair( new CopyBinaryCellFunction() ); //cp is workaround for read bug
+ }
+ else {
+ throw new DMLRuntimeException("Incorrect input format in getRDDHandleForVariable");
+ }
+
+ //keep rdd handle for future operations on it
+ RDDObject rddhandle = new RDDObject(rdd, fo.getVarName());
+ rddhandle.setHDFSFile(true);
+ fo.setRDDHandle(rddhandle);
+ }
+
+ return rdd;
+ }
+
+ /**
* TODO So far we only create broadcast variables but never destroy
* them. This is a memory leak which might lead to executor out-of-memory.
* However, in order to handle this, we need to keep track when broadcast
@@ -888,8 +975,8 @@ public class SparkExecutionContext extends ExecutionContext
public void addLineageRDD(String varParent, String varChild)
throws DMLRuntimeException
{
- RDDObject parent = getMatrixObject(varParent).getRDDHandle();
- RDDObject child = getMatrixObject(varChild).getRDDHandle();
+ RDDObject parent = getCacheableData(varParent).getRDDHandle();
+ RDDObject child = getCacheableData(varChild).getRDDHandle();
parent.addLineageChild( child );
}
@@ -904,8 +991,8 @@ public class SparkExecutionContext extends ExecutionContext
public void addLineageBroadcast(String varParent, String varChild)
throws DMLRuntimeException
{
- RDDObject parent = getMatrixObject(varParent).getRDDHandle();
- BroadcastObject child = getMatrixObject(varChild).getBroadcastHandle();
+ RDDObject parent = getCacheableData(varParent).getRDDHandle();
+ BroadcastObject child = getCacheableData(varChild).getBroadcastHandle();
parent.addLineageChild( child );
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/instructions/MRJobInstruction.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/instructions/MRJobInstruction.java b/src/main/java/org/apache/sysml/runtime/instructions/MRJobInstruction.java
index 2811dd8..67cf2f3 100644
--- a/src/main/java/org/apache/sysml/runtime/instructions/MRJobInstruction.java
+++ b/src/main/java/org/apache/sysml/runtime/instructions/MRJobInstruction.java
@@ -33,6 +33,7 @@ import org.apache.sysml.lops.runtime.RunMRJobs;
import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.runtime.DMLRuntimeException;
import org.apache.sysml.runtime.controlprogram.ParForProgramBlock.PDataPartitionFormat;
+import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysml.runtime.controlprogram.parfor.ProgramConverter;
@@ -1026,6 +1027,14 @@ public class MRJobInstruction extends Instruction
if ( d.getDataType() == DataType.MATRIX ) {
inputmat.add((MatrixObject) d);
}
+ else if( d.getDataType() == DataType.FRAME ) {
+ //FIXME conversion from frame to matrix object (meta data only) to adhere to
+ //the given matrix-based mr job submission framework
+ FrameObject fo = (FrameObject) d;
+ MatrixObject mo = new MatrixObject(fo.getValueType(), fo.getFileName(), fo.getMetaData());
+ mo.setFileFormatProperties(fo.getFileFormatProperties());
+ inputmat.add(mo);
+ }
}
inputMatrices = inputmat.toArray(new MatrixObject[inputmat.size()]);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/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 04a7613..5f6ac5f 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
@@ -25,6 +25,7 @@ import org.apache.sysml.lops.Lop;
import org.apache.sysml.parser.Statement;
import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysml.runtime.functionobjects.ParameterizedBuiltin;
@@ -229,10 +230,10 @@ public class ParameterizedBuiltinCPInstruction extends ComputationCPInstruction
ec.releaseMatrixInput(params.get("target"));
}
else if ( opcode.equalsIgnoreCase("transform")) {
- MatrixObject mo = (MatrixObject) ec.getVariable(params.get("target"));
+ FrameObject fo = (FrameObject) ec.getVariable(params.get("target"));
MatrixObject out = (MatrixObject) ec.getVariable(output.getName());
try {
- JobReturn jt = DataTransform.cpDataTransform(this, new MatrixObject[] { mo } , new MatrixObject[] {out} );
+ JobReturn jt = DataTransform.cpDataTransform(this, new FrameObject[] { fo } , new MatrixObject[] {out} );
out.updateMatrixCharacteristics(jt.getMatrixCharacteristics(0));
} catch (Exception e) {
throw new DMLRuntimeException(e);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/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 da7a8a0..8926bb7 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
@@ -28,6 +28,7 @@ import org.apache.sysml.lops.UnaryCP;
import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysml.runtime.controlprogram.parfor.ProgramConverter;
@@ -278,71 +279,68 @@ public class VariableCPInstruction extends CPInstruction
switch (voc) {
case CreateVariable:
- // variable name (only supports Matrices, and only w/ double value type)
- in1 = new CPOperand(parts[1], ValueType.DOUBLE, DataType.MATRIX);
+ // variable name
+ DataType dt = DataType.valueOf(parts[4]);
+ ValueType vt = dt==DataType.MATRIX ? ValueType.DOUBLE : ValueType.STRING;
+ in1 = new CPOperand(parts[1], vt, dt);
// file name
in2 = new CPOperand(parts[2], ValueType.STRING, DataType.SCALAR);
// file name override flag
in3 = new CPOperand(parts[3], ValueType.BOOLEAN, DataType.SCALAR);
// format
- String fmt = parts[4];
+ String fmt = parts[5];
if ( fmt.equalsIgnoreCase("csv") ) {
- /*
- * Cretevar instructions for CSV format either has 13 or 14 inputs.
- * 13 inputs: createvar corresponding to WRITE -- includes properties hasHeader, delim, and sparse
- * 14 inputs: createvar corresponding to READ -- includes properties hasHeader, delim, fill, and fillValue
- */
- if ( parts.length < 14 || parts.length > 16 )
+ // Cretevar instructions for CSV format either has 13 or 14 inputs.
+ // 13 inputs: createvar corresponding to WRITE -- includes properties hasHeader, delim, and sparse
+ // 14 inputs: createvar corresponding to READ -- includes properties hasHeader, delim, fill, and fillValue
+ if ( parts.length < 15 || parts.length > 17 )
throw new DMLRuntimeException("Invalid number of operands in createvar instruction: " + str);
}
else {
- if ( parts.length != 5 && parts.length != 11 )
+ if ( parts.length != 6 && parts.length != 12 )
throw new DMLRuntimeException("Invalid number of operands in createvar instruction: " + str);
}
OutputInfo oi = OutputInfo.stringToOutputInfo(fmt);
InputInfo ii = OutputInfo.getMatchingInputInfo(oi);
MatrixCharacteristics mc = new MatrixCharacteristics();
- if ( parts.length == 5 ) {
+ if ( parts.length == 6 ) {
// do nothing
- ;
}
- else if ( parts.length >= 10 ) {
+ else if ( parts.length >= 11 ) {
// matrix characteristics
- mc.setDimension(Long.parseLong(parts[5]), Long.parseLong(parts[6]));
- mc.setBlockSize(Integer.parseInt(parts[7]), Integer.parseInt(parts[8]));
- mc.setNonZeros(Long.parseLong(parts[9]));
+ mc.setDimension(Long.parseLong(parts[6]), Long.parseLong(parts[7]));
+ mc.setBlockSize(Integer.parseInt(parts[8]), Integer.parseInt(parts[9]));
+ mc.setNonZeros(Long.parseLong(parts[10]));
}
else {
throw new DMLRuntimeException("Invalid number of operands in createvar instruction: " + str);
}
MatrixFormatMetaData iimd = new MatrixFormatMetaData(mc, oi, ii);
boolean updateInPlace = false;
- if ( parts.length >= 11 )
- updateInPlace = Boolean.parseBoolean(parts[10]);
+ if ( parts.length >= 12 )
+ updateInPlace = Boolean.parseBoolean(parts[11]);
if ( fmt.equalsIgnoreCase("csv") ) {
- /*
- * Cretevar instructions for CSV format either has 13 or 14 inputs.
- * 13 inputs: createvar corresponding to WRITE -- includes properties hasHeader, delim, and sparse
- * 14 inputs: createvar corresponding to READ -- includes properties hasHeader, delim, fill, and fillValue
- */
+ // Cretevar instructions for CSV format either has 13 or 14 inputs.
+ // 13 inputs: createvar corresponding to WRITE -- includes properties hasHeader, delim, and sparse
+ // 14 inputs: createvar corresponding to READ -- includes properties hasHeader, delim, fill, and fillValue
FileFormatProperties fmtProperties = null;
- if ( parts.length == 14 ) {
- boolean hasHeader = Boolean.parseBoolean(parts[11]);
- String delim = parts[12];
- boolean sparse = Boolean.parseBoolean(parts[13]);
+ if ( parts.length == 15 ) {
+ boolean hasHeader = Boolean.parseBoolean(parts[12]);
+ String delim = parts[13];
+ boolean sparse = Boolean.parseBoolean(parts[14]);
fmtProperties = new CSVFileFormatProperties(hasHeader, delim, sparse) ;
}
else {
- boolean hasHeader = Boolean.parseBoolean(parts[11]);
- String delim = parts[12];
- boolean fill = Boolean.parseBoolean(parts[13]);
- double fillValue = UtilFunctions.parseToDouble(parts[14]);
+ boolean hasHeader = Boolean.parseBoolean(parts[12]);
+ String delim = parts[13];
+ boolean fill = Boolean.parseBoolean(parts[14]);
+ double fillValue = UtilFunctions.parseToDouble(parts[15]);
String naStrings = null;
- if ( parts.length == 16 )
- naStrings = parts[15];
+ if ( parts.length == 17 )
+ naStrings = parts[16];
fmtProperties = new CSVFileFormatProperties(hasHeader, delim, fill, fillValue, naStrings) ;
}
return new VariableCPInstruction(VariableOperationCode.CreateVariable, in1, in2, in3, iimd, updateInPlace, parts.length, fmtProperties, opcode, str);
@@ -353,8 +351,6 @@ public class VariableCPInstruction extends CPInstruction
case AssignVariable:
in1 = new CPOperand(parts[1]);
in2 = new CPOperand(parts[2]);
- //if ( in1.getValueType() != in2.getValueType() )
- // throw new DMLRuntimeException("Value type mismatch while assigning variables ("+in1.getValueType()+", "+in2.getValueType()+").");
break;
case CopyVariable:
@@ -456,6 +452,15 @@ public class VariableCPInstruction extends CPInstruction
if(DMLScript.STATISTICS && updateInPlace)
Statistics.incrementTotalUIPVar();
}
+ else if( input1.getDataType() == DataType.FRAME ) {
+ String fname = input2.getName();
+ FrameObject fobj = new FrameObject(fname);
+ fobj.setVarName(input1.getName());
+ fobj.setDataType(DataType.FRAME);
+ fobj.setMetaData((MetaData)metadata.clone());
+ fobj.setFileFormatProperties(formatProperties);
+ ec.setVariable(input1.getName(), fobj);
+ }
else if ( input1.getDataType() == DataType.SCALAR ){
ScalarObject sobj = null;
ec.setScalarOutput(input1.getName(), sobj);
@@ -707,25 +712,26 @@ public class VariableCPInstruction extends CPInstruction
//get filename (literal or variable expression)
String fname = ec.getScalarInput(input2.getName(), ValueType.STRING, input2.isLiteral()).getStringValue();
- if ( input1.getDataType() == DataType.SCALAR ) {
+ if( input1.getDataType() == DataType.SCALAR ) {
writeScalarToHDFS(ec, fname);
}
- else
- {
+ else if( input1.getDataType() == DataType.MATRIX ){
String outFmt = input3.getName();
-
- if (outFmt.equalsIgnoreCase("matrixmarket")) {
+ if (outFmt.equalsIgnoreCase("matrixmarket"))
writeMMFile(ec, fname);
- }
- else if (outFmt.equalsIgnoreCase("csv") ) {
+ else if (outFmt.equalsIgnoreCase("csv") )
writeCSVFile(ec, fname);
- }
else {
// Default behavior
MatrixObject mo = (MatrixObject)ec.getVariable(input1.getName());
mo.exportData(fname, outFmt);
}
}
+ else if( input1.getDataType() == DataType.FRAME ) {
+ String outFmt = input3.getName();
+ FrameObject mo = (FrameObject)ec.getVariable(input1.getName());
+ mo.exportData(fname, outFmt);
+ }
}
/**
@@ -929,7 +935,7 @@ public class VariableCPInstruction extends CPInstruction
return parseInstruction(str);
}
- private static String getBasicCreateVarString(String varName, String fileName, boolean fNameOverride, String format) {
+ private static String getBasicCreateVarString(String varName, String fileName, boolean fNameOverride, DataType dt, String format) {
StringBuilder sb = new StringBuilder();
sb.append("CP");
sb.append(Lop.OPERAND_DELIMITOR);
@@ -942,17 +948,19 @@ public class VariableCPInstruction extends CPInstruction
sb.append(Lop.OPERAND_DELIMITOR);
sb.append(fNameOverride);
sb.append(Lop.OPERAND_DELIMITOR);
+ sb.append(dt.toString());
+ sb.append(Lop.OPERAND_DELIMITOR);
sb.append(format);
return sb.toString();
}
- public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, String format) throws DMLRuntimeException {
- return parseInstruction(getBasicCreateVarString(varName, fileName, fNameOverride, format));
+ public static Instruction prepareCreateMatrixVariableInstruction(String varName, String fileName, boolean fNameOverride, String format) throws DMLRuntimeException {
+ return parseInstruction(getBasicCreateVarString(varName, fileName, fNameOverride, DataType.MATRIX, format));
}
- public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, String format, MatrixCharacteristics mc) throws DMLRuntimeException {
+ public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, DataType dt, String format, MatrixCharacteristics mc) throws DMLRuntimeException {
StringBuilder sb = new StringBuilder();
- sb.append(getBasicCreateVarString(varName, fileName, fNameOverride, format));
+ sb.append(getBasicCreateVarString(varName, fileName, fNameOverride, dt, format));
sb.append(Lop.OPERAND_DELIMITOR);
sb.append(mc.getRows());
@@ -970,9 +978,9 @@ public class VariableCPInstruction extends CPInstruction
return parseInstruction(str);
}
- public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, String format, MatrixCharacteristics mc, boolean updateInPlace) throws DMLRuntimeException {
+ public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, DataType dt, String format, MatrixCharacteristics mc, boolean updateInPlace) throws DMLRuntimeException {
StringBuilder sb = new StringBuilder();
- sb.append(getBasicCreateVarString(varName, fileName, fNameOverride, format));
+ sb.append(getBasicCreateVarString(varName, fileName, fNameOverride, dt, format));
sb.append(Lop.OPERAND_DELIMITOR);
sb.append(mc.getRows());
@@ -992,9 +1000,9 @@ public class VariableCPInstruction extends CPInstruction
return parseInstruction(str);
}
- public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, String format, MatrixCharacteristics mc, boolean updateInPlace, boolean hasHeader, String delim, boolean sparse) throws DMLRuntimeException {
+ public static Instruction prepareCreateVariableInstruction(String varName, String fileName, boolean fNameOverride, DataType dt, String format, MatrixCharacteristics mc, boolean updateInPlace, boolean hasHeader, String delim, boolean sparse) throws DMLRuntimeException {
StringBuilder sb = new StringBuilder();
- sb.append(getBasicCreateVarString(varName, fileName, fNameOverride, format));
+ sb.append(getBasicCreateVarString(varName, fileName, fNameOverride, dt, format));
sb.append(Lop.OPERAND_DELIMITOR);
sb.append(mc.getRows());
@@ -1054,6 +1062,7 @@ public class VariableCPInstruction extends CPInstruction
{
return ( opcode == VariableOperationCode.CastAsScalarVariable ||
opcode == VariableOperationCode.CastAsMatrixVariable ||
+ opcode == VariableOperationCode.CastAsFrameVariable ||
opcode == VariableOperationCode.CastAsIntegerVariable ||
opcode == VariableOperationCode.CastAsDoubleVariable ||
opcode == VariableOperationCode.CastAsBooleanVariable );
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/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 782161f..b8d2f6c 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
@@ -33,6 +33,7 @@ import org.apache.sysml.lops.PartialAggregate.CorrectionLocationType;
import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.parser.Statement;
import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysml.runtime.controlprogram.context.SparkExecutionContext;
@@ -386,7 +387,7 @@ public class ParameterizedBuiltinSPInstruction extends ComputationSPInstruction
try {
DataTransform.spDataTransform(
this,
- new MatrixObject[] { (MatrixObject) sec.getVariable(params.get("target")) },
+ new FrameObject[] { (FrameObject) sec.getVariable(params.get("target")) },
new MatrixObject[] { (MatrixObject) sec.getVariable(output.getName()) }, ec);
} catch (Exception e) {
throw new DMLRuntimeException(e);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/io/FrameReaderFactory.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/io/FrameReaderFactory.java b/src/main/java/org/apache/sysml/runtime/io/FrameReaderFactory.java
index e18a7d2..35ec5f7 100644
--- a/src/main/java/org/apache/sysml/runtime/io/FrameReaderFactory.java
+++ b/src/main/java/org/apache/sysml/runtime/io/FrameReaderFactory.java
@@ -103,10 +103,6 @@ public class FrameReaderFactory
public static FrameReader createFrameReader( InputInfo iinfo, FileFormatProperties props )
throws DMLRuntimeException
{
- //check valid read properties
- if( props == null )
- throw new DMLRuntimeException("Failed to create frame reader with empty properties.");
-
FrameReader reader = null;
if( iinfo == InputInfo.TextCellInputInfo ) {
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java b/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java
index a88923f..4b3b58e 100644
--- a/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java
+++ b/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java
@@ -56,6 +56,8 @@ import org.apache.sysml.parser.Expression.DataType;
import org.apache.sysml.parser.Expression.ValueType;
import org.apache.sysml.parser.ParameterizedBuiltinFunctionExpression;
import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.controlprogram.caching.CacheableData;
+import org.apache.sysml.runtime.controlprogram.caching.FrameObject;
import org.apache.sysml.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysml.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysml.runtime.controlprogram.context.SparkExecutionContext;
@@ -713,17 +715,17 @@ public class DataTransform
* @return
* @throws Exception
*/
- public static JobReturn mrDataTransform(MRJobInstruction jobinst, MatrixObject[] inputMatrices, String shuffleInst, String otherInst, byte[] resultIndices, MatrixObject[] outputMatrices, int numReducers, int replication) throws Exception {
+ public static JobReturn mrDataTransform(MRJobInstruction jobinst, MatrixObject[] inputs, String shuffleInst, String otherInst, byte[] resultIndices, MatrixObject[] outputs, int numReducers, int replication) throws Exception {
String[] insts = shuffleInst.split(Instruction.INSTRUCTION_DELIM);
// Parse transform instruction (the first instruction) to obtain relevant fields
- TransformOperands oprnds = new TransformOperands(insts[0], inputMatrices[0]);
+ TransformOperands oprnds = new TransformOperands(insts[0], inputs[0]);
JobConf job = new JobConf(ConfigurationManager.getCachedJobConf());
FileSystem fs = FileSystem.get(job);
- // find the first file in alphabetical ordering of partfiles in directory inputPath
+ // find the first file in alphabetical ordering of part files in directory inputPath
String smallestFile = CSVReblockMR.findSmallestFile(job, oprnds.inputPath);
// find column names
@@ -739,10 +741,10 @@ public class DataTransform
ArrayList<Integer> bboutputs = new ArrayList<Integer>();
// divide output objects based on output format (CSV or BinaryBlock)
- for(int i=0; i < outputMatrices.length; i++)
+ for(int i=0; i < outputs.length; i++)
{
- if(outputMatrices[i].getFileFormatProperties() != null
- && outputMatrices[i].getFileFormatProperties().getFileFormat() == FileFormatProperties.FileFormat.CSV)
+ if(outputs[i].getFileFormatProperties() != null
+ && outputs[i].getFileFormatProperties().getFileFormat() == FileFormatProperties.FileFormat.CSV)
csvoutputs.add(i);
else
bboutputs.add(i);
@@ -751,7 +753,7 @@ public class DataTransform
boolean isBB = (bboutputs.size() > 0);
String tmpPath = MRJobConfiguration.constructTempOutputFilename();
- checkIfOutputOverlapsWithTxMtd(outputMatrices, oprnds, isCSV, isBB, csvoutputs, bboutputs, fs);
+ checkIfOutputOverlapsWithTxMtd(outputs, oprnds, isCSV, isBB, csvoutputs, bboutputs, fs);
JobReturn retCSV = null, retBB = null;
@@ -779,7 +781,7 @@ public class DataTransform
// Apply transformation metadata, and perform actual transformation
if(isCSV)
retCSV = ApplyTfCSVMR.runJob(oprnds.inputPath, specWithIDs, oprnds.txMtdPath, tmpPath,
- outputMatrices[csvoutputs.get(0)].getFileName(), partOffsetsFile,
+ outputs[csvoutputs.get(0)].getFileName(), partOffsetsFile,
oprnds.inputCSVProperties, numColumns, replication, outHeader);
if(isBB)
@@ -796,7 +798,7 @@ public class DataTransform
throw new DMLRuntimeException(ERROR_MSG_ZERO_ROWS);
retBB = ApplyTfBBMR.runJob(oprnds.inputPath, insts[1], otherInst,
- specWithIDs, oprnds.txMtdPath, tmpPath, outputMatrices[bboutputs.get(0)].getFileName(),
+ specWithIDs, oprnds.txMtdPath, tmpPath, outputs[bboutputs.get(0)].getFileName(),
ret1.counterFile.toString(), oprnds.inputCSVProperties, numRowsTf, numColumns,
numColumnsTf, replication, outHeader);
}
@@ -835,7 +837,7 @@ public class DataTransform
// Apply transformation metadata, and perform actual transformation
retCSV = ApplyTfCSVMR.runJob(oprnds.inputPath, specWithIDs, oprnds.applyTxPath, tmpPath,
- outputMatrices[csvoutputs.get(0)].getFileName(), ret1.counterFile.toString(),
+ outputs[csvoutputs.get(0)].getFileName(), ret1.counterFile.toString(),
oprnds.inputCSVProperties, numColumns, replication, outHeader);
}
@@ -855,7 +857,7 @@ public class DataTransform
// apply transformation metadata, as well as reblock the resulting data
retBB = ApplyTfBBMR.runJob(oprnds.inputPath, insts[1], otherInst, specWithIDs,
- oprnds.txMtdPath, tmpPath, outputMatrices[bboutputs.get(0)].getFileName(),
+ oprnds.txMtdPath, tmpPath, outputs[bboutputs.get(0)].getFileName(),
ret1.counterFile.toString(), oprnds.inputCSVProperties, ret1.rlens[0],
ret1.clens[0], numColumnsTf, replication, outHeader);
}
@@ -874,7 +876,7 @@ public class DataTransform
oprnds.inputCSVProperties.getDelim(), // use the same header as the input
false, Double.NaN, null);
- MapReduceTool.writeMetaDataFile (outputMatrices[csvoutputs.get(0)].getFileName()+".mtd",
+ MapReduceTool.writeMetaDataFile (outputs[csvoutputs.get(0)].getFileName()+".mtd",
ValueType.DOUBLE, retCSV.getMatrixCharacteristics(0),
OutputInfo.CSVOutputInfo, prop);
return retCSV;
@@ -884,7 +886,7 @@ public class DataTransform
{
retBB.getMatrixCharacteristics(0).setDimension(numRowsTf, numColumnsTf);
- MapReduceTool.writeMetaDataFile (outputMatrices[bboutputs.get(0)].getFileName()+".mtd",
+ MapReduceTool.writeMetaDataFile (outputs[bboutputs.get(0)].getFileName()+".mtd",
ValueType.DOUBLE, retBB.getMatrixCharacteristics(0), OutputInfo.BinaryBlockOutputInfo);
return retBB;
}
@@ -968,20 +970,24 @@ public class DataTransform
* @throws JSONException
* @throws IllegalArgumentException
*/
- public static JobReturn cpDataTransform(ParameterizedBuiltinCPInstruction inst, MatrixObject[] inputMatrices, MatrixObject[] outputMatrices) throws IOException, DMLRuntimeException, IllegalArgumentException, JSONException {
- TransformOperands oprnds = new TransformOperands(inst.getParameterMap(), inputMatrices[0]);
- return cpDataTransform(oprnds, inputMatrices, outputMatrices);
+ public static JobReturn cpDataTransform(ParameterizedBuiltinCPInstruction inst, CacheableData<?>[] inputs, MatrixObject[] outputs) throws IOException, DMLRuntimeException, IllegalArgumentException, JSONException {
+ TransformOperands oprnds = new TransformOperands(inst.getParameterMap(), inputs[0]);
+ return cpDataTransform(oprnds, inputs, outputs);
}
- public static JobReturn cpDataTransform(String inst, MatrixObject[] inputMatrices, MatrixObject[] outputMatrices) throws IOException, DMLRuntimeException, IllegalArgumentException, JSONException {
+ public static JobReturn cpDataTransform(String inst, CacheableData<?>[] inputs, MatrixObject[] outputs)
+ throws IOException, DMLRuntimeException, IllegalArgumentException, JSONException
+ {
String[] insts = inst.split(Instruction.INSTRUCTION_DELIM);
// Parse transform instruction (the first instruction) to obtain relevant fields
- TransformOperands oprnds = new TransformOperands(insts[0], inputMatrices[0]);
+ TransformOperands oprnds = new TransformOperands(insts[0], inputs[0]);
- return cpDataTransform(oprnds, inputMatrices, outputMatrices);
+ return cpDataTransform(oprnds, inputs, outputs);
}
- public static JobReturn cpDataTransform(TransformOperands oprnds, MatrixObject[] inputMatrices, MatrixObject[] outputMatrices) throws IOException, DMLRuntimeException, IllegalArgumentException, JSONException {
+ public static JobReturn cpDataTransform(TransformOperands oprnds, CacheableData<?>[] inputs, MatrixObject[] outputs)
+ throws IOException, DMLRuntimeException, IllegalArgumentException, JSONException
+ {
JobConf job = new JobConf(ConfigurationManager.getCachedJobConf());
FileSystem fs = FileSystem.get(job);
// find the first file in alphabetical ordering of partfiles in directory inputPath
@@ -996,9 +1002,9 @@ public class DataTransform
ArrayList<Integer> bboutputs = new ArrayList<Integer>();
// divide output objects based on output format (CSV or BinaryBlock)
- for(int i=0; i < outputMatrices.length; i++)
+ for(int i=0; i < outputs.length; i++)
{
- if(outputMatrices[i].getFileFormatProperties() != null && outputMatrices[i].getFileFormatProperties().getFileFormat() == FileFormatProperties.FileFormat.CSV)
+ if(outputs[i].getFileFormatProperties() != null && outputs[i].getFileFormatProperties().getFileFormat() == FileFormatProperties.FileFormat.CSV)
csvoutputs.add(i);
else
bboutputs.add(i);
@@ -1006,16 +1012,18 @@ public class DataTransform
boolean isCSV = (csvoutputs.size() > 0);
boolean isBB = (bboutputs.size() > 0);
- checkIfOutputOverlapsWithTxMtd(outputMatrices, oprnds, isCSV, isBB, csvoutputs, bboutputs, fs);
+ checkIfOutputOverlapsWithTxMtd(outputs, oprnds, isCSV, isBB, csvoutputs, bboutputs, fs);
JobReturn ret = null;
if (!oprnds.isApply) {
// build specification file with column IDs insteadof column names
- String specWithIDs = processSpecFile(fs, oprnds.inputPath, smallestFile, colNamesToIds, oprnds.inputCSVProperties, oprnds.spec);
+ String specWithIDs = processSpecFile(fs, oprnds.inputPath, smallestFile, colNamesToIds, oprnds.inputCSVProperties,
+ oprnds.spec);
MapReduceTool.writeStringToHDFS(specWithIDs, oprnds.txMtdPath + "/" + "spec.json");
- ret = performTransform(job, fs, oprnds.inputPath, colNamesToIds.size(), oprnds.inputCSVProperties, specWithIDs, oprnds.txMtdPath, oprnds.isApply, outputMatrices[0], outHeader, isBB, isCSV );
+ ret = performTransform(job, fs, oprnds.inputPath, colNamesToIds.size(), oprnds.inputCSVProperties, specWithIDs,
+ oprnds.txMtdPath, oprnds.isApply, outputs[0], outHeader, isBB, isCSV );
}
else {
// copy given transform metadata (applyTxPath) to specified location (txMtdPath)
@@ -1026,7 +1034,8 @@ public class DataTransform
String specWithIDs = (oprnds.spec != null) ?
oprnds.spec : MapReduceTool.readStringFromHDFSFile(oprnds.txMtdPath + "/" + "spec.json");
- ret = performTransform(job, fs, oprnds.inputPath, colNamesToIds.size(), oprnds.inputCSVProperties, specWithIDs, oprnds.txMtdPath, oprnds.isApply, outputMatrices[0], outHeader, isBB, isCSV );
+ ret = performTransform(job, fs, oprnds.inputPath, colNamesToIds.size(), oprnds.inputCSVProperties, specWithIDs,
+ oprnds.txMtdPath, oprnds.isApply, outputs[0], outHeader, isBB, isCSV );
}
return ret;
@@ -1351,13 +1360,13 @@ public class DataTransform
br.close();
}
- private static void checkIfOutputOverlapsWithTxMtd(MatrixObject[] outputMatrices, TransformOperands oprnds,
+ private static void checkIfOutputOverlapsWithTxMtd(MatrixObject[] outputs, TransformOperands oprnds,
boolean isCSV, boolean isBB, ArrayList<Integer> csvoutputs, ArrayList<Integer> bboutputs, FileSystem fs) throws DMLRuntimeException {
if(isCSV) {
- checkIfOutputOverlapsWithTxMtd(oprnds.txMtdPath, outputMatrices[csvoutputs.get(0)].getFileName(), fs);
+ checkIfOutputOverlapsWithTxMtd(oprnds.txMtdPath, outputs[csvoutputs.get(0)].getFileName(), fs);
}
else if(isBB) {
- checkIfOutputOverlapsWithTxMtd(oprnds.txMtdPath, outputMatrices[bboutputs.get(0)].getFileName(), fs);
+ checkIfOutputOverlapsWithTxMtd(oprnds.txMtdPath, outputs[bboutputs.get(0)].getFileName(), fs);
}
}
@@ -1384,17 +1393,17 @@ public class DataTransform
}
}
- public static void spDataTransform(ParameterizedBuiltinSPInstruction inst, MatrixObject[] inputMatrices, MatrixObject[] outputMatrices, ExecutionContext ec) throws Exception {
+ public static void spDataTransform(ParameterizedBuiltinSPInstruction inst, FrameObject[] inputs, MatrixObject[] outputs, ExecutionContext ec) throws Exception {
SparkExecutionContext sec = (SparkExecutionContext)ec;
// Parse transform instruction (the first instruction) to obtain relevant fields
- TransformOperands oprnds = new TransformOperands(inst.getParams(), inputMatrices[0]);
+ TransformOperands oprnds = new TransformOperands(inst.getParams(), inputs[0]);
JobConf job = new JobConf();
FileSystem fs = FileSystem.get(job);
- checkIfOutputOverlapsWithTxMtd(oprnds.txMtdPath, outputMatrices[0].getFileName(), fs);
+ checkIfOutputOverlapsWithTxMtd(oprnds.txMtdPath, outputs[0].getFileName(), fs);
// find the first file in alphabetical ordering of partfiles in directory inputPath
String smallestFile = CSVReblockMR.findSmallestFile(job, oprnds.inputPath);
@@ -1409,7 +1418,7 @@ public class DataTransform
// Construct RDD for input data
@SuppressWarnings("unchecked")
- JavaPairRDD<LongWritable, Text> inputData = (JavaPairRDD<LongWritable, Text>) sec.getRDDHandleForMatrixObject(inputMatrices[0], InputInfo.CSVInputInfo);
+ JavaPairRDD<LongWritable, Text> inputData = (JavaPairRDD<LongWritable, Text>) sec.getRDDHandleForFrameObject(inputs[0], InputInfo.CSVInputInfo);
JavaRDD<Tuple2<LongWritable,Text>> csvLines = JavaPairRDD.toRDD(inputData).toJavaRDD();
long numRowsTf=0, numColumnsTf=0;
@@ -1469,7 +1478,7 @@ public class DataTransform
if ( outtfPairRDD != null )
{
- MatrixObject outMO = outputMatrices[0];
+ MatrixObject outMO = outputs[0];
String outVar = outMO.getVarName();
outMO.setRDDHandle(new RDDObject(outtfPairRDD, outVar));
sec.addLineageRDD(outVar, inst.getParams().get("target"));
@@ -1495,9 +1504,9 @@ public class DataTransform
private boolean isApply=false;
private CSVFileFormatProperties inputCSVProperties = null;
- private TransformOperands(String inst, MatrixObject inputMatrix) {
- inputPath = inputMatrix.getFileName();
- inputCSVProperties = (CSVFileFormatProperties)inputMatrix.getFileFormatProperties();
+ private TransformOperands(String inst, CacheableData<?> input) {
+ inputPath = input.getFileName();
+ inputCSVProperties = (CSVFileFormatProperties)input.getFileFormatProperties();
String[] instParts = inst.split(Instruction.OPERAND_DELIM);
txMtdPath = instParts[3];
applyTxPath = instParts[4].startsWith("applymtd=") ? instParts[4].substring(9) : null;
@@ -1509,14 +1518,14 @@ public class DataTransform
outNamesFile = instParts[pos].startsWith("outnames=") ? instParts[pos].substring(9) : null;
}
- private TransformOperands(HashMap<String, String> params, MatrixObject inputMatrix) {
- inputPath = inputMatrix.getFileName();
+ private TransformOperands(HashMap<String, String> params, CacheableData<?> input) {
+ inputPath = input.getFileName();
txMtdPath = params.get(ParameterizedBuiltinFunctionExpression.TF_FN_PARAM_MTD);
spec = params.get(ParameterizedBuiltinFunctionExpression.TF_FN_PARAM_SPEC);
applyTxPath = params.get(ParameterizedBuiltinFunctionExpression.TF_FN_PARAM_APPLYMTD);
isApply = (applyTxPath != null);
outNamesFile = params.get(ParameterizedBuiltinFunctionExpression.TF_FN_PARAM_OUTNAMES); // can be null
- inputCSVProperties = (CSVFileFormatProperties)inputMatrix.getFileFormatProperties();
+ inputCSVProperties = (CSVFileFormatProperties)input.getFileFormatProperties();
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/test/java/org/apache/sysml/test/integration/functions/transform/TransformEncodeDecodeTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/sysml/test/integration/functions/transform/TransformEncodeDecodeTest.java b/src/test/java/org/apache/sysml/test/integration/functions/transform/TransformEncodeDecodeTest.java
new file mode 100644
index 0000000..551e529
--- /dev/null
+++ b/src/test/java/org/apache/sysml/test/integration/functions/transform/TransformEncodeDecodeTest.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysml.test.integration.functions.transform;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.apache.sysml.api.DMLScript.RUNTIME_PLATFORM;
+import org.apache.sysml.lops.LopProperties.ExecType;
+import org.apache.sysml.runtime.DMLRuntimeException;
+import org.apache.sysml.runtime.io.FrameReader;
+import org.apache.sysml.runtime.io.FrameReaderFactory;
+import org.apache.sysml.runtime.io.FrameWriter;
+import org.apache.sysml.runtime.io.FrameWriterFactory;
+import org.apache.sysml.runtime.matrix.data.FrameBlock;
+import org.apache.sysml.runtime.matrix.data.InputInfo;
+import org.apache.sysml.runtime.matrix.data.OutputInfo;
+import org.apache.sysml.runtime.util.DataConverter;
+import org.apache.sysml.test.integration.AutomatedTestBase;
+import org.apache.sysml.test.integration.TestConfiguration;
+import org.apache.sysml.test.utils.TestUtils;
+
+/**
+ *
+ */
+public class TransformEncodeDecodeTest extends AutomatedTestBase
+{
+ private static final String TEST_NAME1 = "TransformEncodeDecode";
+ private static final String TEST_DIR = "functions/transform/";
+ private static final String TEST_CLASS_DIR = TEST_DIR + TransformEncodeDecodeTest.class.getSimpleName() + "/";
+
+ private static final String SPEC = "TransformEncodeDecodeSpec.json";
+
+ private static final int rows = 1234;
+ private static final int cols = 2;
+ private static final double sparsity1 = 0.9;
+ private static final double sparsity2 = 0.1;
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ addTestConfiguration(TEST_NAME1, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1,new String[]{"FO"}));
+ }
+
+ @Test
+ public void runTestCSVDenseCP() {
+ runTransformEncodeDecodeTest(ExecType.CP, false, "csv");
+ }
+
+ @Test
+ public void runTestCSVSparseCP() {
+ runTransformEncodeDecodeTest(ExecType.CP, true, "csv");
+ }
+
+ @Test
+ public void runTestTextcellDenseCP() {
+ runTransformEncodeDecodeTest(ExecType.CP, false, "text");
+ }
+
+ @Test
+ public void runTestTextcellSparseCP() {
+ runTransformEncodeDecodeTest(ExecType.CP, true, "text");
+ }
+
+ /*TODO to enable binary, we need to properly propagate schema first
+ @Test
+ public void runTestBinaryDenseCP() {
+ runTransformEncodeDecodeTest(ExecType.CP, false, "binary");
+ }
+
+ @Test
+ public void runTestBinarySparseCP() {
+ runTransformEncodeDecodeTest(ExecType.CP, true, "binary");
+ }
+ */
+
+
+ /**
+ *
+ * @param sparseM1
+ * @param sparseM2
+ * @param instType
+ * @throws IOException
+ * @throws DMLRuntimeException
+ */
+ private void runTransformEncodeDecodeTest( ExecType et, boolean sparse, String fmt)
+ {
+ RUNTIME_PLATFORM platformOld = rtplatform;
+ rtplatform = RUNTIME_PLATFORM.HYBRID; //only CP supported
+
+ try
+ {
+ getAndLoadTestConfiguration(TEST_NAME1);
+
+ //get input/output info
+ InputInfo iinfo = InputInfo.stringExternalToInputInfo(fmt);
+ OutputInfo oinfo = InputInfo.getMatchingOutputInfo(iinfo);
+
+ //generate and write input data
+ double[][] A = TestUtils.round(getRandomMatrix(rows, cols, 1, 15, sparse ? sparsity2 : sparsity1, 7));
+ FrameBlock FA = DataConverter.convertToFrameBlock(DataConverter.convertToMatrixBlock(A));
+ FrameWriter writer = FrameWriterFactory.createFrameWriter(oinfo);
+ writer.writeFrameToHDFS(FA, input("F"), rows, cols);
+
+ fullDMLScriptName = SCRIPT_DIR+TEST_DIR+TEST_NAME1 + ".dml";
+ programArgs = new String[]{"-explain","-args", input("F"), fmt,
+ String.valueOf(rows), String.valueOf(cols),
+ SCRIPT_DIR+TEST_DIR+SPEC, output("FO") };
+
+ //run test
+ runTest(true, false, null, -1);
+
+ //compare matrices (values recoded to identical codes)
+ FrameReader reader = FrameReaderFactory.createFrameReader(iinfo);
+ FrameBlock FO = reader.readFrameFromHDFS(output("FO"), 16, 2);
+ HashMap<String,Long> cFA = getCounts(FA, 1);
+ Iterator<String[]> iterFO = FO.getStringRowIterator();
+ while( iterFO.hasNext() ) {
+ String[] row = iterFO.next();
+ Double expected = (double)cFA.get(row[1]);
+ Double val = (row[0]!=null)?Double.valueOf(row[0]) : 0;
+ Assert.assertEquals("Output aggregates don't match: "+
+ expected+" vs "+val, expected, val);
+ }
+ }
+ catch(Exception ex) {
+ ex.printStackTrace();
+ Assert.fail(ex.getMessage());
+ }
+ finally {
+ rtplatform = platformOld;
+ }
+ }
+
+ /**
+ *
+ * @param input
+ * @param groupBy
+ * @return
+ */
+ private HashMap<String, Long> getCounts(FrameBlock input, int groupBy) {
+ HashMap<String, Long> ret = new HashMap<String, Long>();
+ for( int i=0; i<input.getNumRows(); i++ ) {
+ String key = input.get(i, groupBy).toString();
+ Long tmp = ret.get(key);
+ ret.put(key, (tmp!=null)?tmp+1:1);
+ }
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/test/scripts/functions/transform/TransformEncodeDecode.dml
----------------------------------------------------------------------
diff --git a/src/test/scripts/functions/transform/TransformEncodeDecode.dml b/src/test/scripts/functions/transform/TransformEncodeDecode.dml
new file mode 100644
index 0000000..b2b9e2a
--- /dev/null
+++ b/src/test/scripts/functions/transform/TransformEncodeDecode.dml
@@ -0,0 +1,33 @@
+#-------------------------------------------------------------
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+#-------------------------------------------------------------
+
+F1 = read($1, data_type="frame", format=$2, rows=$3, cols=$4);
+jspec = read($5, data_type="scalar", value_type="string");
+
+[X, M] = transformencode(target=F1, spec=jspec);
+
+A = aggregate(target=X[,1], groups=X[,2], fn="count");
+Ag = append(A, seq(1,nrow(A)));
+
+F2 = transformdecode(target=Ag, spec=jspec, meta=M);
+
+write(F2, $6, format=$2);
+
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/test/scripts/functions/transform/TransformEncodeDecodeSpec.json
----------------------------------------------------------------------
diff --git a/src/test/scripts/functions/transform/TransformEncodeDecodeSpec.json b/src/test/scripts/functions/transform/TransformEncodeDecodeSpec.json
new file mode 100644
index 0000000..2ec7e6f
--- /dev/null
+++ b/src/test/scripts/functions/transform/TransformEncodeDecodeSpec.json
@@ -0,0 +1,5 @@
+{
+ "ids": true
+ ,"recode": [ 2 ]
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/test_suites/java/org/apache/sysml/test/integration/functions/frame/ZPackageSuite.java
----------------------------------------------------------------------
diff --git a/src/test_suites/java/org/apache/sysml/test/integration/functions/frame/ZPackageSuite.java b/src/test_suites/java/org/apache/sysml/test/integration/functions/frame/ZPackageSuite.java
index 353d181..37e54ec 100644
--- a/src/test_suites/java/org/apache/sysml/test/integration/functions/frame/ZPackageSuite.java
+++ b/src/test_suites/java/org/apache/sysml/test/integration/functions/frame/ZPackageSuite.java
@@ -31,6 +31,7 @@ import org.junit.runners.Suite;
FrameCopyTest.class,
FrameGetSetTest.class,
FrameIndexingTest.class,
+ FrameReadWriteTest.class,
FrameSerializationTest.class,
})
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/test_suites/java/org/apache/sysml/test/integration/functions/jmlc/ZPackageSuite.java
----------------------------------------------------------------------
diff --git a/src/test_suites/java/org/apache/sysml/test/integration/functions/jmlc/ZPackageSuite.java b/src/test_suites/java/org/apache/sysml/test/integration/functions/jmlc/ZPackageSuite.java
index 742a140..e643aa8 100644
--- a/src/test_suites/java/org/apache/sysml/test/integration/functions/jmlc/ZPackageSuite.java
+++ b/src/test_suites/java/org/apache/sysml/test/integration/functions/jmlc/ZPackageSuite.java
@@ -26,7 +26,11 @@ import org.junit.runners.Suite;
* won't run two of them at once. */
@RunWith(Suite.class)
@Suite.SuiteClasses({
+ FrameCastingTest.class,
FrameDecodeTest.class,
+ FrameEncodeTest.class,
+ FrameIndexingAppendTest.class,
+ FrameLeftIndexingTest.class,
FrameReadMetaTest.class,
FrameTransformTest.class,
ReuseModelVariablesTest.class,
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7528146c/src/test_suites/java/org/apache/sysml/test/integration/functions/transform/ZPackageSuite.java
----------------------------------------------------------------------
diff --git a/src/test_suites/java/org/apache/sysml/test/integration/functions/transform/ZPackageSuite.java b/src/test_suites/java/org/apache/sysml/test/integration/functions/transform/ZPackageSuite.java
index 999fb64..0bbe7f3 100644
--- a/src/test_suites/java/org/apache/sysml/test/integration/functions/transform/ZPackageSuite.java
+++ b/src/test_suites/java/org/apache/sysml/test/integration/functions/transform/ZPackageSuite.java
@@ -29,6 +29,7 @@ import org.junit.runners.Suite;
RunTest.class,
ScalingTest.class,
TransformAndApplyTest.class,
+ TransformEncodeDecodeTest.class,
TransformTest.class,
})