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/02/14 23:15:56 UTC
[1/3] incubator-systemml git commit: Fix mr iqm pick range record
reader (current part weights), incl cleanup
Repository: incubator-systemml
Updated Branches:
refs/heads/master ce9392741 -> 7f8716b64
Fix mr iqm pick range record reader (current part weights), incl cleanup
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/2fa69cd6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/2fa69cd6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/2fa69cd6
Branch: refs/heads/master
Commit: 2fa69cd668e76e90baabfb2dcfdf7c549f6b3a2f
Parents: ce93927
Author: Matthias Boehm <mb...@us.ibm.com>
Authored: Sat Feb 13 17:46:58 2016 -0800
Committer: Matthias Boehm <mb...@us.ibm.com>
Committed: Sat Feb 13 17:46:58 2016 -0800
----------------------------------------------------------------------
.../matrix/sort/PickFromCompactInputFormat.java | 191 +++++++------------
1 file changed, 72 insertions(+), 119 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/2fa69cd6/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java b/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
index 7e8aef0..d03afc8 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
@@ -48,8 +48,7 @@ import org.apache.sysml.runtime.matrix.data.Pair;
//key class to read has to be DoubleWritable
public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, MatrixCell>
-{
-
+{
public static final String VALUE_IS_WEIGHT="value.is.weight";
public static final String INPUT_IS_VECTOR="input.is.vector";
public static final String SELECTED_RANGES="selected.ranges";
@@ -191,11 +190,9 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
wt += counts[partID];
}
sb.append(partID+"," + (wt-counts[partID]) + ";");
-
sb.append(sumwt + "," + lbound + "," + ubound);
- //System.out.println("range string: " + sb.toString());
+
job.set(SELECTED_RANGES, sb.toString());
-
job.setBoolean(INPUT_IS_VECTOR, false);
}
@@ -222,111 +219,89 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
public static class RangePickRecordReader implements RecordReader<MatrixIndexes, MatrixCell>
{
- //private boolean valueIsWeight=true;
- protected long totLength;
- protected FileSystem fs;
- protected Path path;
- protected JobConf conf;
+ private Path path;
- protected FSDataInputStream currentStream;
-
+ private FSDataInputStream currentStream;
private int startPos=0;
private int numToRead=0;
- DoubleWritable readKey=new DoubleWritable();
- Writable readValue = new IntWritable(0);
+ private DoubleWritable readKey=new DoubleWritable();
+ private Writable readValue = new IntWritable(0);
private boolean noRecordsNeeded=false;
private int rawKeyValuesRead=0;
private int index=0;
- //private int currentRepeat=0;
+ private int beginPart=-1, endPart=-1, currPart=-1;
+ private double sumwt = 0.0, readWt; // total weight (set in JobConf)
+ private double lbound, ubound;
+ private HashMap<Integer,Double> partWeights = null;
+ private boolean isFirstRecord = true;
+ private ReadWithZeros reader=null;
- int beginPart=-1, endPart=-1, currPart=-1;
- double sumwt = 0.0, readWt, wtUntilCurrPart; // total weight (set in JobConf)
- double lbound, ubound;
- double[] partWeights = null;
- boolean isFirstRecord = true;
-
-
- //to handle zeros
- ReadWithZeros reader=null;
- /*private boolean contain0s=false;
- private boolean justFound0=false;
- private DoubleWritable keyAfterZero;
- private Writable valueAfterZero;
- private long numZeros=0;*/
-
- private int getIndexInTheArray(String name)
- {
- int i=name.indexOf("part-");
- assert(i>=0);
- return Integer.parseInt(name.substring(i+5));
- }
-
- private void parseSelectedRangeString(String str) {
- String[] f1 = str.split(";");
- String[] f2 = null;
-
- // Each field of the form: "pid,wt" where wt is the total wt until the part pid
- partWeights = new double[f1.length-1];
- for(int i=0; i < f1.length-1; i++) {
- f2 = f1[i].split(",");
- if(i==0) {
- beginPart = Integer.parseInt(f2[0]);
- }
- if (i==f1.length-2) {
- endPart = Integer.parseInt(f2[0]);
- }
- partWeights[i] = Double.parseDouble(f2[1]);
- }
-
- // last field: "sumwt, lbound, ubound"
- f2 = f1[f1.length-1].split(",");
- sumwt = Double.parseDouble(f2[0]);
- lbound = Double.parseDouble(f2[1]);
- ubound = Double.parseDouble(f2[2]);
- }
-
@SuppressWarnings("unchecked")
- public RangePickRecordReader(JobConf job, FileSplit split) throws IOException {
+ public RangePickRecordReader(JobConf job, FileSplit split)
+ throws IOException
+ {
parseSelectedRangeString(job.get(SELECTED_RANGES));
// check if the current part file needs to be processed
path = split.getPath();
- totLength = split.getLength();
currentStream = FileSystem.get(job).open(path);
currPart = getIndexInTheArray(path.getName());
- //System.out.println("RangePickRecordReader(): sumwt " + sumwt + " currPart " + currPart + " partRange [" + beginPart + "," + endPart + "]");
if ( currPart < beginPart || currPart > endPart ) {
- System.out.println(" currPart is out of range. Skipping part!");
noRecordsNeeded = true;
return;
}
Class<? extends Writable> valueClass=(Class<? extends Writable>) job.getClass(VALUE_CLASS, Writable.class);
try {
- // readKey=keyClass.newInstance();
readValue=valueClass.newInstance();
} catch (Exception e) {
throw new RuntimeException(e);
}
- //valueIsWeight=job.getBoolean(VALUE_IS_WEIGHT, true);
int part0=job.getInt(PARTITION_OF_ZERO, -1);
boolean contain0s=false;
long numZeros =0;
- if(part0==currPart)
- {
+ if(part0==currPart) {
contain0s = true;
numZeros = job.getLong(NUMBER_OF_ZERO, 0);
}
reader=new ReadWithZeros(currentStream, contain0s, numZeros);
}
- public boolean next(MatrixIndexes key, MatrixCell value) throws IOException {
- assert(currPart!=-1);
+ private int getIndexInTheArray(String name) {
+ int i=name.indexOf("part-");
+ return Integer.parseInt(name.substring(i+5));
+ }
+
+ private void parseSelectedRangeString(String str) {
+ String[] f1 = str.split(";");
+ String[] f2 = null;
+ // Each field of the form: "pid,wt" where wt is the total wt until the part pid
+ partWeights = new HashMap<Integer,Double>();
+ for(int i=0; i < f1.length-1; i++) {
+ f2 = f1[i].split(",");
+ if(i==0) {
+ beginPart = Integer.parseInt(f2[0]);
+ }
+ if (i==f1.length-2) {
+ endPart = Integer.parseInt(f2[0]);
+ }
+ partWeights.put(i, Double.parseDouble(f2[1]));
+ }
+
+ // last field: "sumwt, lbound, ubound"
+ f2 = f1[f1.length-1].split(",");
+ sumwt = Double.parseDouble(f2[0]);
+ lbound = Double.parseDouble(f2[1]);
+ ubound = Double.parseDouble(f2[2]);
+ }
+
+ public boolean next(MatrixIndexes key, MatrixCell value)
+ throws IOException
+ {
if(noRecordsNeeded)
- // this part file does not fall within the required range of values
return false;
double qLowerD = sumwt*lbound; // lower quantile in double
@@ -334,15 +309,14 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
// weight until current part
if (isFirstRecord) {
- readWt = partWeights[currPart];
- wtUntilCurrPart = partWeights[currPart];
+ if( !partWeights.containsKey(currPart) )
+ return false; //noRecordsNeeded
+ readWt = partWeights.get(currPart);
isFirstRecord = false;
}
double tmpWt = 0;
if ( currPart == beginPart || currPart == endPart ) {
- //readWt = partWeights[currPart];
-
reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
tmpWt = ((IntWritable)readValue).get();
@@ -356,14 +330,13 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
key.setIndexes(++index,1);
value.setValue(readKey.get()*tmpWt);
readWt += tmpWt;
- //System.out.println("currpart " + currPart + ", return (" + index +",1): " + readKey.get()*tmpWt + "| [" + readKey.get() + "," + tmpWt +"] readWt=" + readWt);
return true;
}
else {
return false;
}
}
- else { //if(currPart != beginPart && currPart != endPart) {
+ else {
// read full part
reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
tmpWt = ((IntWritable)readValue).get();
@@ -372,7 +345,6 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
value.setValue(readKey.get()*tmpWt);
readWt += tmpWt;
- //System.out.println("currpart " + currPart + ", return (" + index +",1): " + readKey.get()*tmpWt + "| [" + readKey.get() + "," + tmpWt +"] readWt=" + readWt);
return true;
}
}
@@ -408,68 +380,58 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
}
}
+ /**
+ *
+ */
public static class PickRecordReader implements RecordReader<MatrixIndexes, MatrixCell>
{
private boolean valueIsWeight=true;
- protected long totLength;
- protected FileSystem fs;
- protected Path path;
- protected JobConf conf;
-
- protected FSDataInputStream currentStream;
+ private FileSystem fs;
+ private Path path;
+
+ private FSDataInputStream currentStream;
private int posIndex=0;
private int[] pos=null; //starting from 0
private int[] indexes=null;
- DoubleWritable readKey=new DoubleWritable();
- Writable readValue;
+ private DoubleWritable readKey=new DoubleWritable();
+ private Writable readValue;
private int numRead=0;
private boolean noRecordsNeeded=false;
-
- //to handle zeros
ReadWithZeros reader=null;
- /*private boolean contain0s=false;
- private boolean justFound0=false;
- private DoubleWritable keyAfterZero;
- private Writable valueAfterZero;
- private long numZeros=0;*/
private int getIndexInTheArray(String name)
{
int i=name.indexOf("part-");
- assert(i>=0);
return Integer.parseInt(name.substring(i+5));
}
@SuppressWarnings("unchecked")
public PickRecordReader(JobConf job, FileSplit split)
- throws IOException{
+ throws IOException
+ {
fs = FileSystem.get(job);
path = split.getPath();
- totLength = split.getLength();
currentStream = fs.open(path);
int partIndex=getIndexInTheArray(path.getName());
String arrStr=job.get(SELECTED_POINTS_PREFIX+partIndex);
- //System.out.println("read back in the recordreader: "+arrStr);
- if(arrStr==null || arrStr.isEmpty())
- {
+ if(arrStr==null || arrStr.isEmpty()) {
noRecordsNeeded=true;
return;
}
+
String[] strs=arrStr.split(",");
pos=new int[strs.length];
indexes=new int[strs.length];
- for(int i=0; i<strs.length; i++)
- {
+ for(int i=0; i<strs.length; i++) {
String[] temp=strs[i].split(":");
pos[i]=Integer.parseInt(temp[0]);
indexes[i]=Integer.parseInt(temp[1]);
}
- // Class<? extends WritableComparable> keyClass=(Class<? extends WritableComparable>) job.getClass(KEY_CLASS, WritableComparable.class);
+
Class<? extends Writable> valueClass=(Class<? extends Writable>) job.getClass(VALUE_CLASS, Writable.class);
try {
- // readKey=keyClass.newInstance();
- readValue=valueClass.newInstance();
+ readValue=valueClass.newInstance();
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -478,8 +440,7 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
int part0=job.getInt(PARTITION_OF_ZERO, -1);
boolean contain0s=false;
long numZeros =0;
- if(part0==partIndex)
- {
+ if(part0==partIndex) {
contain0s = true;
numZeros = job.getLong(NUMBER_OF_ZERO, 0);
}
@@ -487,21 +448,15 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
}
public boolean next(MatrixIndexes key, MatrixCell value)
- throws IOException {
-
- if(noRecordsNeeded || posIndex>=pos.length)
- {
- //System.out.println("return false");
- //System.out.println("noRecordsNeeded="+noRecordsNeeded+", currentStream.getPos()="+currentStream.getPos()
- // +", totLength="+totLength+", posIndex="+posIndex+", pos.length="+pos.length);
+ throws IOException
+ {
+ if(noRecordsNeeded || posIndex>=pos.length) {
return false;
}
- //System.out.println("numRead="+numRead+" pos["+posIndex+"]="+pos[posIndex]);
while(numRead<=pos[posIndex])
{
reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
- // System.out.println("**** numRead "+numRead+" -- "+readKey+": "+readValue);
if(valueIsWeight)
numRead+=((IntWritable)readValue).get();
else
@@ -511,14 +466,12 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
key.setIndexes(indexes[posIndex], 1);
value.setValue(readKey.get());
posIndex++;
- //System.out.println("next: "+key+", "+value);
return true;
}
@Override
public void close() throws IOException {
- currentStream.close();
-
+ currentStream.close();
}
@Override
[2/3] incubator-systemml git commit: Fix mr iqm/quantile record
readers (bounded partition read, progress)
Posted by mb...@apache.org.
Fix mr iqm/quantile record readers (bounded partition read, progress)
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/240143bb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/240143bb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/240143bb
Branch: refs/heads/master
Commit: 240143bbad57489872b86114cee4ac206fa9c2a4
Parents: 2fa69cd
Author: Matthias Boehm <mb...@us.ibm.com>
Authored: Sat Feb 13 22:16:30 2016 -0800
Committer: Matthias Boehm <mb...@us.ibm.com>
Committed: Sat Feb 13 22:16:30 2016 -0800
----------------------------------------------------------------------
.../matrix/sort/PickFromCompactInputFormat.java | 53 +++++++-------------
.../runtime/matrix/sort/ReadWithZeros.java | 13 +++--
2 files changed, 25 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/240143bb/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java b/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
index d03afc8..7dd3c13 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/sort/PickFromCompactInputFormat.java
@@ -222,12 +222,10 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
private Path path;
private FSDataInputStream currentStream;
- private int startPos=0;
- private int numToRead=0;
+ protected long totLength;
private DoubleWritable readKey=new DoubleWritable();
- private Writable readValue = new IntWritable(0);
+ private IntWritable readValue = new IntWritable(0);
private boolean noRecordsNeeded=false;
- private int rawKeyValuesRead=0;
private int index=0;
private int beginPart=-1, endPart=-1, currPart=-1;
private double sumwt = 0.0, readWt; // total weight (set in JobConf)
@@ -236,7 +234,6 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
private boolean isFirstRecord = true;
private ReadWithZeros reader=null;
- @SuppressWarnings("unchecked")
public RangePickRecordReader(JobConf job, FileSplit split)
throws IOException
{
@@ -244,6 +241,7 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
// check if the current part file needs to be processed
path = split.getPath();
+ totLength = split.getLength();
currentStream = FileSystem.get(job).open(path);
currPart = getIndexInTheArray(path.getName());
@@ -252,13 +250,6 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
return;
}
- Class<? extends Writable> valueClass=(Class<? extends Writable>) job.getClass(VALUE_CLASS, Writable.class);
- try {
- readValue=valueClass.newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
int part0=job.getInt(PARTITION_OF_ZERO, -1);
boolean contain0s=false;
long numZeros =0;
@@ -317,20 +308,20 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
double tmpWt = 0;
if ( currPart == beginPart || currPart == endPart ) {
- reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
- tmpWt = ((IntWritable)readValue).get();
+ boolean ret = reader.readNextKeyValuePairs(readKey, readValue);
+ tmpWt = readValue.get();
while(readWt+tmpWt < qLowerD) {
readWt += tmpWt;
- reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
- tmpWt = ((IntWritable)readValue).get();
+ ret &= reader.readNextKeyValuePairs(readKey, readValue);
+ tmpWt = readValue.get();
}
if((readWt<qLowerD && readWt+tmpWt >= qLowerD) || (readWt+tmpWt <= qUpperD) || (readWt<qUpperD && readWt+tmpWt>=qUpperD)) {
key.setIndexes(++index,1);
value.setValue(readKey.get()*tmpWt);
readWt += tmpWt;
- return true;
+ return ret;
}
else {
return false;
@@ -338,20 +329,19 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
}
else {
// read full part
- reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
- tmpWt = ((IntWritable)readValue).get();
+ boolean ret = reader.readNextKeyValuePairs(readKey, readValue);
+ tmpWt = readValue.get();
key.setIndexes(++index,1);
value.setValue(readKey.get()*tmpWt);
readWt += tmpWt;
- return true;
+ return ret;
}
}
@Override
public void close() throws IOException {
- //DO Nothing
currentStream.close();
}
@@ -373,10 +363,8 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
@Override
public float getProgress() throws IOException {
- if(numToRead>0)
- return (float)(rawKeyValuesRead-startPos)/(float)numToRead;
- else
- return 100.0f;
+ float progress = (float) getPos() / totLength;
+ return (progress>=0 && progress<=1) ? progress : 1.0f;
}
}
@@ -394,8 +382,8 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
private int[] pos=null; //starting from 0
private int[] indexes=null;
- private DoubleWritable readKey=new DoubleWritable();
- private Writable readValue;
+ private DoubleWritable readKey = new DoubleWritable();
+ private IntWritable readValue = new IntWritable();
private int numRead=0;
private boolean noRecordsNeeded=false;
ReadWithZeros reader=null;
@@ -406,7 +394,6 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
return Integer.parseInt(name.substring(i+5));
}
- @SuppressWarnings("unchecked")
public PickRecordReader(JobConf job, FileSplit split)
throws IOException
{
@@ -429,12 +416,6 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
indexes[i]=Integer.parseInt(temp[1]);
}
- Class<? extends Writable> valueClass=(Class<? extends Writable>) job.getClass(VALUE_CLASS, Writable.class);
- try {
- readValue=valueClass.newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
valueIsWeight=job.getBoolean(VALUE_IS_WEIGHT, true);
int part0=job.getInt(PARTITION_OF_ZERO, -1);
@@ -456,9 +437,9 @@ public class PickFromCompactInputFormat extends FileInputFormat<MatrixIndexes, M
while(numRead<=pos[posIndex])
{
- reader.readNextKeyValuePairs(readKey, (IntWritable)readValue);
+ reader.readNextKeyValuePairs(readKey, readValue);
if(valueIsWeight)
- numRead+=((IntWritable)readValue).get();
+ numRead+=readValue.get();
else
numRead++;
}
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/240143bb/src/main/java/org/apache/sysml/runtime/matrix/sort/ReadWithZeros.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/sort/ReadWithZeros.java b/src/main/java/org/apache/sysml/runtime/matrix/sort/ReadWithZeros.java
index dc578a8..8793be9 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/sort/ReadWithZeros.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/sort/ReadWithZeros.java
@@ -28,9 +28,7 @@ import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
public class ReadWithZeros
-{
-
-
+{
private boolean contain0s=false;
private long numZeros=0;
private FSDataInputStream currentStream;
@@ -46,8 +44,11 @@ public class ReadWithZeros
numZeros=num0;
}
- public void readNextKeyValuePairs(DoubleWritable readKey, IntWritable readValue)throws IOException
+ public boolean readNextKeyValuePairs(DoubleWritable readKey, IntWritable readValue)
+ throws IOException
{
+ boolean ret = true;
+
try {
if(contain0s && justFound0)
{
@@ -68,7 +69,7 @@ public class ReadWithZeros
readValue.set((int)numZeros);
}
else {
- throw e;
+ ret = false;
}
}
@@ -80,5 +81,7 @@ public class ReadWithZeros
readKey.set(0);
readValue.set((int)numZeros);
}
+
+ return ret;
}
}
[3/3] incubator-systemml git commit: Performance mr
iqm/quantile/median (qsort num red, qpick buffer size)
Posted by mb...@apache.org.
Performance mr iqm/quantile/median (qsort num red, qpick buffer size)
Our mr iqm/quantile/median realization via qsort and qpick showed severe
scalability bottlenecks on large data. The root cause was that the
out-of-core qpick reads -- according to the required quantiles -- one or
two partitions into CP, where the partition sizes are determined by the
number of reducers of SortMR. This performance patch makes two
improvements: (1) setting the number of reducers for SortMR (except for
order) such that a single partition has at most 10M records (~128MB)
which bounds the read for qpick, and (2) a larger io buffer size of 64KB
for qpick which makes the partition read slightly faster compared to the
4KB default (~20%).
Project: http://git-wip-us.apache.org/repos/asf/incubator-systemml/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-systemml/commit/7f8716b6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-systemml/tree/7f8716b6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-systemml/diff/7f8716b6
Branch: refs/heads/master
Commit: 7f8716b641a45dbc77713710979cbf331d89ae8a
Parents: 240143b
Author: Matthias Boehm <mb...@us.ibm.com>
Authored: Sun Feb 14 01:53:42 2016 -0800
Committer: Matthias Boehm <mb...@us.ibm.com>
Committed: Sun Feb 14 01:53:42 2016 -0800
----------------------------------------------------------------------
.../org/apache/sysml/runtime/matrix/SortMR.java | 16 +++++++++++-----
.../apache/sysml/runtime/util/MapReduceTool.java | 3 ++-
2 files changed, 13 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7f8716b6/src/main/java/org/apache/sysml/runtime/matrix/SortMR.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/matrix/SortMR.java b/src/main/java/org/apache/sysml/runtime/matrix/SortMR.java
index 218a95b..6502e57 100644
--- a/src/main/java/org/apache/sysml/runtime/matrix/SortMR.java
+++ b/src/main/java/org/apache/sysml/runtime/matrix/SortMR.java
@@ -196,13 +196,19 @@ public class SortMR
Path outpath = new Path(tmpOutput);
FileOutputFormat.setOutputPath(job, outpath);
MapReduceTool.deleteFileIfExistOnHDFS(outpath, job);
-
+
//set number of reducers (1 if local mode)
- if( InfrastructureAnalyzer.isLocalMode(job) )
- job.setNumReduceTasks(1);
- else
+ if( !InfrastructureAnalyzer.isLocalMode(job) ) {
MRJobConfiguration.setNumReducers(job, numReducers, numReducers);
-
+ //ensure partition size <= 10M records to avoid scalability bottlenecks
+ //on cp-side qpick instructions for quantile/iqm/median (~128MB)
+ if( !(getSortInstructionType(sortInst)==SortKeys.OperationTypes.Indexes) )
+ job.setNumReduceTasks((int)Math.max(job.getNumReduceTasks(), rlen/10000000));
+ }
+ else //in case of local mode
+ job.setNumReduceTasks(1);
+
+
//setup input/output format
job.setInputFormat(SamplingSortMRInputFormat.class);
SamplingSortMRInputFormat.setTargetKeyValueClasses(job, (Class<? extends WritableComparable>) outputInfo.outputKeyClass, outputInfo.outputValueClass);
http://git-wip-us.apache.org/repos/asf/incubator-systemml/blob/7f8716b6/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java b/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
index a142a2d..9b70106 100644
--- a/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
+++ b/src/main/java/org/apache/sysml/runtime/util/MapReduceTool.java
@@ -620,7 +620,8 @@ public class MapReduceTool
if(fileToRead==null)
throw new RuntimeException("cannot read partition "+currentPart);
- FSDataInputStream currentStream=fs.open(fileToRead);
+ int buffsz = 64 * 1024;
+ FSDataInputStream currentStream=fs.open(fileToRead, buffsz);
DoubleWritable readKey=new DoubleWritable();
IntWritable readValue=new IntWritable();