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/03/31 18:56:47 UTC
[2/2] incubator-systemml git commit: [SYSTEMML-583] Generalization
and cleanup read util transform meta data
[SYSTEMML-583] Generalization and cleanup read util transform meta data
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/25706494
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/25706494
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/25706494
Branch: refs/heads/master
Commit: 25706494fb4c48ad287bc83a2d323a5d833edd17
Parents: 6ffbf95
Author: Matthias Boehm <mb...@us.ibm.com>
Authored: Thu Mar 31 01:30:20 2016 -0700
Committer: Matthias Boehm <mb...@us.ibm.com>
Committed: Thu Mar 31 09:56:11 2016 -0700
----------------------------------------------------------------------
.../org/apache/sysml/api/jmlc/Connection.java | 29 ++++++++++++++------
.../sysml/runtime/transform/DataTransform.java | 22 +++++++--------
.../sysml/runtime/transform/DummycodeAgent.java | 10 ++++---
.../sysml/runtime/transform/RecodeAgent.java | 6 ++--
.../apache/sysml/runtime/transform/TfUtils.java | 11 +++++++-
.../runtime/transform/TransformationAgent.java | 5 ----
6 files changed, 50 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/25706494/src/main/java/org/apache/sysml/api/jmlc/Connection.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/api/jmlc/Connection.java b/src/main/java/org/apache/sysml/api/jmlc/Connection.java
index c5e733b..abb22ed 100644
--- a/src/main/java/org/apache/sysml/api/jmlc/Connection.java
+++ b/src/main/java/org/apache/sysml/api/jmlc/Connection.java
@@ -62,6 +62,7 @@ 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.Pair;
+import org.apache.sysml.runtime.transform.TfUtils;
import org.apache.sysml.runtime.transform.TransformationAgent;
import org.apache.sysml.runtime.transform.TransformationAgent.TX_METHOD;
import org.apache.sysml.runtime.transform.decode.DecoderRecode;
@@ -386,25 +387,35 @@ public class Connection
return ret;
}
-
+ /**
+ *
+ * @param spec
+ * @param metapath
+ * @return
+ * @throws IOException
+ */
+ public FrameBlock readTransformMetaData(String spec, String metapath) throws IOException {
+ return readTransformMetaData(spec, metapath, TfUtils.TXMTD_SEP);
+ }
/**
+ * NOTE: This is a temporary api - use readTransformMetaData(String spec, String metapath) if possible.
*
* @param spec
* @param metapath
* @return
* @throws IOException
*/
- public FrameBlock readTransformMetaData(String spec, String metapath)
+ public FrameBlock readTransformMetaData(String spec, String metapath, String colDelim)
throws IOException
{
//read column types (for sanity check column names)
- String coltypesStr = MapReduceTool.readStringFromHDFSFile(metapath+File.separator+"coltypes.csv");
- List<String> coltypes = Arrays.asList(IOUtilFunctions.split(coltypesStr.trim(), ","));
+ String coltypesStr = MapReduceTool.readStringFromHDFSFile(metapath+File.separator+TfUtils.TXMTD_COLTYPES);
+ List<String> coltypes = Arrays.asList(IOUtilFunctions.split(coltypesStr.trim(), TfUtils.TXMTD_SEP));
//read column names
- String colnamesStr = MapReduceTool.readStringFromHDFSFile(metapath+File.separator+"column.names");
- List<String> colnames = Arrays.asList(IOUtilFunctions.split(colnamesStr.trim(), ","));
+ String colnamesStr = MapReduceTool.readStringFromHDFSFile(metapath+File.separator+TfUtils.TXMTD_COLNAMES);
+ List<String> colnames = Arrays.asList(IOUtilFunctions.split(colnamesStr.trim(), colDelim));
if( coltypes.size() != colnames.size() ) {
LOG.warn("Number of columns names: "+colnames.size()+" (expected: "+coltypes.size()+").");
LOG.warn("--Sample column names: "+(!colnames.isEmpty()?colnames.get(0):"null"));
@@ -416,9 +427,9 @@ public class Connection
for( int j=0; j<colnames.size(); j++ ) {
String colName = colnames.get(j);
String name = metapath+File.separator+"Recode"+File.separator+colName;
- if( MapReduceTool.existsFileOnHDFS(name+".map") ) {
- meta.put(colName, MapReduceTool.readStringFromHDFSFile(name+".map"));
- String ndistinct = MapReduceTool.readStringFromHDFSFile(name+".ndistinct");
+ if( MapReduceTool.existsFileOnHDFS(name+TfUtils.TXMTD_RCD_MAP_SUFFIX) ) {
+ meta.put(colName, MapReduceTool.readStringFromHDFSFile(name+TfUtils.TXMTD_RCD_MAP_SUFFIX));
+ String ndistinct = MapReduceTool.readStringFromHDFSFile(name+TfUtils.TXMTD_RCD_DISTINCT_SUFFIX);
rows = Math.max(rows, Integer.parseInt(ndistinct));
}
else if( coltypes.get(j).equals("2") ) {
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/25706494/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 d47e57c..16f6228 100644
--- a/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java
+++ b/src/main/java/org/apache/sysml/runtime/transform/DataTransform.java
@@ -631,9 +631,9 @@ public class DataTransform
private static void moveFilesFromTmp(FileSystem fs, String tmpPath, String txMtdPath) throws IllegalArgumentException, IOException
{
// move files from temporary location to txMtdPath
- MapReduceTool.renameFileOnHDFS(tmpPath + "/" + TransformationAgent.OUT_HEADER, txMtdPath + "/" + TransformationAgent.OUT_HEADER);
- MapReduceTool.renameFileOnHDFS(tmpPath + "/" + TransformationAgent.OUT_DCD_HEADER, txMtdPath + "/" + TransformationAgent.OUT_DCD_HEADER);
- MapReduceTool.renameFileOnHDFS(tmpPath + "/" + TransformationAgent.COLTYPES_FILE_NAME, txMtdPath + "/" + TransformationAgent.COLTYPES_FILE_NAME);
+ MapReduceTool.renameFileOnHDFS(tmpPath + "/" + TfUtils.TXMTD_COLNAMES, txMtdPath + "/" + TfUtils.TXMTD_COLNAMES);
+ MapReduceTool.renameFileOnHDFS(tmpPath + "/" + TfUtils.TXMTD_DC_COLNAMES, txMtdPath + "/" + TfUtils.TXMTD_DC_COLNAMES);
+ MapReduceTool.renameFileOnHDFS(tmpPath + "/" + TfUtils.TXMTD_COLTYPES, txMtdPath + "/" + TfUtils.TXMTD_COLTYPES);
if ( fs.exists(new Path(tmpPath +"/Dummycode/" + TransformationAgent.DCD_FILE_NAME)) )
{
@@ -677,7 +677,7 @@ public class DataTransform
int id = UtilFunctions.toInt(o);
Path binpath = new Path( tfMtdPath + "/Bin/" + UtilFunctions.unquote(columnNames[id-1]) + TransformationAgent.BIN_FILE_SUFFIX);
- Path rcdpath = new Path( tfMtdPath + "/Recode/" + UtilFunctions.unquote(columnNames[id-1]) + TransformationAgent.NDISTINCT_FILE_SUFFIX);
+ Path rcdpath = new Path( tfMtdPath + "/Recode/" + UtilFunctions.unquote(columnNames[id-1]) + TfUtils.TXMTD_RCD_DISTINCT_SUFFIX);
if ( TfUtils.checkValidInputFile(fs, binpath, false ) )
{
@@ -939,7 +939,7 @@ public class DataTransform
if(oprnds.isApply)
{
- BufferedReader br = new BufferedReader(new InputStreamReader( fs.open(new Path(oprnds.applyTxPath + "/" + TransformationAgent.OUT_HEADER)) ));
+ BufferedReader br = new BufferedReader(new InputStreamReader( fs.open(new Path(oprnds.applyTxPath + "/" + TfUtils.TXMTD_COLNAMES)) ));
ret = br.readLine();
br.close();
}
@@ -1280,14 +1280,14 @@ public class DataTransform
br = new BufferedReader(new InputStreamReader(fs.open(files.get(fileNo))));
if ( fileNo == 0 )
{
- String header = null;
if ( prop.hasHeader() )
br.readLine(); // ignore the header line from data file
- header = headerLine;
- String dcdHeader = _da.constructDummycodedHeader(header, agents.getDelim());
+ //TODO: fix hard-wired header propagation to meta data column names
+
+ String dcdHeader = _da.constructDummycodedHeader(headerLine, agents.getDelim());
numColumnsTf = _da.genDcdMapsAndColTypes(fs, tfMtdPath, ncols, agents);
- DataTransform.generateHeaderFiles(fs, tfMtdPath, header, dcdHeader);
+ generateHeaderFiles(fs, tfMtdPath, headerLine, dcdHeader);
}
line = null;
@@ -1340,13 +1340,13 @@ public class DataTransform
public static void generateHeaderFiles(FileSystem fs, String txMtdDir, String origHeader, String newHeader) throws IOException {
// write out given header line
- Path pt=new Path(txMtdDir+"/" + TransformationAgent.OUT_HEADER);
+ Path pt=new Path(txMtdDir+"/" + TfUtils.TXMTD_COLNAMES);
BufferedWriter br=new BufferedWriter(new OutputStreamWriter(fs.create(pt,true)));
br.write(origHeader+"\n");
br.close();
// write out the new header line (after all transformations)
- pt = new Path(txMtdDir + "/" + TransformationAgent.OUT_DCD_HEADER);
+ pt = new Path(txMtdDir + "/" + TfUtils.TXMTD_DC_COLNAMES);
br=new BufferedWriter(new OutputStreamWriter(fs.create(pt,true)));
br.write(newHeader+"\n");
br.close();
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/25706494/src/main/java/org/apache/sysml/runtime/transform/DummycodeAgent.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/transform/DummycodeAgent.java b/src/main/java/org/apache/sysml/runtime/transform/DummycodeAgent.java
index 079ad58..e623ab5 100644
--- a/src/main/java/org/apache/sysml/runtime/transform/DummycodeAgent.java
+++ b/src/main/java/org/apache/sysml/runtime/transform/DummycodeAgent.java
@@ -20,6 +20,7 @@
package org.apache.sysml.runtime.transform;
import java.io.BufferedWriter;
+import java.io.File;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Arrays;
@@ -39,6 +40,7 @@ import org.apache.wink.json4j.JSONObject;
import com.google.common.base.Functions;
import com.google.common.collect.Ordering;
+
import org.apache.sysml.runtime.util.UtilFunctions;
public class DummycodeAgent extends TransformationAgent {
@@ -150,7 +152,7 @@ public class DummycodeAgent extends TransformationAgent {
{
if ( _dcdList != null && idx < _dcdList.length && _dcdList[idx] == colID )
{
- br.write(colID + "," + "1" + "," + sum + "," + (sum+_domainSizes[idx]-1) + "\n");
+ br.write(colID + TfUtils.TXMTD_SEP + "1" + TfUtils.TXMTD_SEP + sum + TfUtils.TXMTD_SEP + (sum+_domainSizes[idx]-1) + "\n");
_dcdColumnMap[colID-1] = (sum+_domainSizes[idx]-1)-1;
for(int i=sum; i <=(sum+_domainSizes[idx]-1); i++)
@@ -161,7 +163,7 @@ public class DummycodeAgent extends TransformationAgent {
}
else
{
- br.write(colID + "," + "0" + "," + sum + "," + sum + "\n");
+ br.write(colID + TfUtils.TXMTD_SEP + "0" + TfUtils.TXMTD_SEP + sum + TfUtils.TXMTD_SEP + sum + "\n");
_dcdColumnMap[colID-1] = sum-1;
if ( agents.getBinAgent().isBinned(colID) != -1 )
@@ -176,12 +178,12 @@ public class DummycodeAgent extends TransformationAgent {
br.close();
// Write coltypes.csv
- pt=new Path(txMtdDir+"/" + COLTYPES_FILE_NAME);
+ pt=new Path(txMtdDir + File.separator + TfUtils.TXMTD_COLTYPES);
br=new BufferedWriter(new OutputStreamWriter(fs.create(pt,true)));
br.write(columnTypeToID(ctypes[0]) + "");
for(int i = 1; i < _dummycodedLength; i++)
- br.write( "," + columnTypeToID(ctypes[i]));
+ br.write( TfUtils.TXMTD_SEP + columnTypeToID(ctypes[i]));
br.close();
return sum-1;
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/25706494/src/main/java/org/apache/sysml/runtime/transform/RecodeAgent.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/transform/RecodeAgent.java b/src/main/java/org/apache/sysml/runtime/transform/RecodeAgent.java
index 10da62e..aa7c22e 100644
--- a/src/main/java/org/apache/sysml/runtime/transform/RecodeAgent.java
+++ b/src/main/java/org/apache/sysml/runtime/transform/RecodeAgent.java
@@ -244,7 +244,7 @@ public class RecodeAgent extends TransformationAgent {
boolean isRecoded = (isRecoded(colID) != -1);
boolean isModeImputed = (mvagent.getMethod(colID) == MVMethod.GLOBAL_MODE);
- Path pt=new Path(outputDir+"/Recode/"+ agents.getName(colID) + RCD_MAP_FILE_SUFFIX);
+ Path pt=new Path(outputDir+"/Recode/"+ agents.getName(colID) + TfUtils.TXMTD_RCD_MAP_SUFFIX);
BufferedWriter br=null;
if(isRecoded)
br = new BufferedWriter(new OutputStreamWriter(fs.create(pt,true)));
@@ -299,7 +299,7 @@ public class RecodeAgent extends TransformationAgent {
br.close();
// output number of distinct values
- pt=new Path(outputDir+"/Recode/"+ agents.getName(colID) + NDISTINCT_FILE_SUFFIX);
+ pt=new Path(outputDir+"/Recode/"+ agents.getName(colID) + TfUtils.TXMTD_RCD_DISTINCT_SUFFIX);
br=new BufferedWriter(new OutputStreamWriter(fs.create(pt,true)));
br.write(""+map.size());
br.close();
@@ -382,7 +382,7 @@ public class RecodeAgent extends TransformationAgent {
for(int i=0; i<_rcdList.length;i++) {
int colID = _rcdList[i];
- Path path = new Path( txMtdDir + "/Recode/" + agents.getName(colID) + RCD_MAP_FILE_SUFFIX);
+ Path path = new Path( txMtdDir + "/Recode/" + agents.getName(colID) + TfUtils.TXMTD_RCD_MAP_SUFFIX);
TfUtils.checkValidInputFile(fs, path, true);
HashMap<String,String> map = new HashMap<String,String>();
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/25706494/src/main/java/org/apache/sysml/runtime/transform/TfUtils.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/transform/TfUtils.java b/src/main/java/org/apache/sysml/runtime/transform/TfUtils.java
index 75c53e3..bc08a3a 100644
--- a/src/main/java/org/apache/sysml/runtime/transform/TfUtils.java
+++ b/src/main/java/org/apache/sysml/runtime/transform/TfUtils.java
@@ -53,7 +53,12 @@ public class TfUtils implements Serializable{
private static final long serialVersionUID = 526252850872633125L;
- public static final String TXMTD_SEP = ",";
+ public static final String TXMTD_SEP = ",";
+ public static final String TXMTD_COLTYPES = "coltypes.csv";
+ public static final String TXMTD_COLNAMES = "column.names";
+ public static final String TXMTD_DC_COLNAMES = "dummycoded.column.names";
+ public static final String TXMTD_RCD_MAP_SUFFIX = ".map";
+ public static final String TXMTD_RCD_DISTINCT_SUFFIX = ".ndistinct";
private OmitAgent _oa = null;
@@ -224,6 +229,8 @@ public class TfUtils implements Serializable{
_numTransformedRows = 0;
_numTransformedColumns = 0;
+ //TODO: fix hard-wired header propagation to meta data column names
+
_headerLine = headerLine;
_hasHeader = hasHeader;
_delimString = delim;
@@ -374,6 +381,8 @@ public class TfUtils implements Serializable{
public String processHeaderLine() throws IOException
{
+ //TODO: fix hard-wired header propagation to meta data column names
+
FileSystem fs = FileSystem.get(ConfigurationManager.getCachedJobConf());
String dcdHeader = getDummycodeAgent().constructDummycodedHeader(getHeader(), getDelim());
getDummycodeAgent().genDcdMapsAndColTypes(fs, getTmpDir(), (int) getNumCols(), this);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/25706494/src/main/java/org/apache/sysml/runtime/transform/TransformationAgent.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/transform/TransformationAgent.java b/src/main/java/org/apache/sysml/runtime/transform/TransformationAgent.java
index be877fa..0cdf682 100644
--- a/src/main/java/org/apache/sysml/runtime/transform/TransformationAgent.java
+++ b/src/main/java/org/apache/sysml/runtime/transform/TransformationAgent.java
@@ -57,18 +57,13 @@ public abstract class TransformationAgent implements Serializable {
public static final String JSON_NBINS = "numbins";
protected static final String MV_FILE_SUFFIX = ".impute";
- protected static final String RCD_MAP_FILE_SUFFIX = ".map";
- protected static final String NDISTINCT_FILE_SUFFIX = ".ndistinct";
protected static final String MODE_FILE_SUFFIX = ".mode";
protected static final String BIN_FILE_SUFFIX = ".bin";
protected static final String SCALE_FILE_SUFFIX = ".scale";
protected static final String DCD_FILE_NAME = "dummyCodeMaps.csv";
- protected static final String COLTYPES_FILE_NAME = "coltypes.csv";
protected static final String DCD_NAME_SEP = "_";
- protected static final String OUT_HEADER = "column.names";
- protected static final String OUT_DCD_HEADER = "dummycoded.column.names";
abstract public void print();
abstract public void mapOutputTransformationMetadata(OutputCollector<IntWritable, DistinctValue> out, int taskID, TfUtils agents) throws IOException;