You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hama.apache.org by ed...@apache.org on 2009/10/22 09:49:48 UTC
svn commit: r828342 - in /incubator/hama/trunk: ./
src/java/org/apache/hama/matrix/ src/java/org/apache/hama/matrix/algebra/
src/test/org/apache/hama/matrix/
Author: edwardyoon
Date: Thu Oct 22 07:49:48 2009
New Revision: 828342
URL: http://svn.apache.org/viewvc?rev=828342&view=rev
Log:
Replacement of TransposeMap/Reduce
Modified:
incubator/hama/trunk/CHANGES.txt
incubator/hama/trunk/src/java/org/apache/hama/matrix/AbstractMatrix.java
incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeMap.java
incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeReduce.java
incubator/hama/trunk/src/test/org/apache/hama/matrix/TestAbstractMatrix.java
Modified: incubator/hama/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/CHANGES.txt?rev=828342&r1=828341&r2=828342&view=diff
==============================================================================
--- incubator/hama/trunk/CHANGES.txt (original)
+++ incubator/hama/trunk/CHANGES.txt Thu Oct 22 07:49:48 2009
@@ -41,6 +41,7 @@
IMPROVEMENTS
+ HAMA-204: Replacement of TransposeMap/Reduce (edwardyoon)
HAMA-213: Replacement of TableReadMapper (edwardyoon)
HAMA-203: Replacement of RandomMatrixMap/Reduce (edwardyoon)
HAMA-197: Replacement for deprecated API of Hbase (edwardyoon)
Modified: incubator/hama/trunk/src/java/org/apache/hama/matrix/AbstractMatrix.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/matrix/AbstractMatrix.java?rev=828342&r1=828341&r2=828342&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/matrix/AbstractMatrix.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/matrix/AbstractMatrix.java Thu Oct 22 07:49:48 2009
@@ -586,17 +586,23 @@
result = new DenseMatrix(config, this.getRows(), this.getColumns());
}
- JobConf jobConf = new JobConf(config);
- jobConf.setJobName("transpose MR job" + result.getPath());
+ Job job = new Job(config, "set MR job : " + this.getPath());
- jobConf.setNumMapTasks(config.getNumMapTasks());
- jobConf.setNumReduceTasks(config.getNumReduceTasks());
-
- TransposeMap.initJob(this.getPath(), TransposeMap.class, IntWritable.class,
- MapWritable.class, jobConf);
- TransposeReduce.initJob(result.getPath(), TransposeReduce.class, jobConf);
+ Scan scan = new Scan();
+ scan.addFamily(Bytes.toBytes(Constants.COLUMN_FAMILY));
+ org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.initTableMapperJob(
+ this.getPath(), scan, TransposeMap.class, IntWritable.class,
+ MapWritable.class, job);
+ org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.initTableReducerJob(
+ result.getPath(), TransposeReduce.class, job);
+ try {
+ job.waitForCompletion(true);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } catch (ClassNotFoundException e) {
+ e.printStackTrace();
+ }
- JobManager.execute(jobConf);
return result;
}
Modified: incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeMap.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeMap.java?rev=828342&r1=828341&r2=828342&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeMap.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeMap.java Thu Oct 22 07:49:48 2009
@@ -22,45 +22,27 @@
import java.io.IOException;
import java.util.Map;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapred.FileInputFormat;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reporter;
-import org.apache.hama.Constants;
-import org.apache.hama.mapred.VectorInputFormat;
+import org.apache.hama.matrix.DenseVector;
+import org.apache.hama.util.BytesUtil;
-public class TransposeMap extends MapReduceBase implements
- Mapper<IntWritable, MapWritable, IntWritable, MapWritable> {
+public class TransposeMap extends TableMapper<IntWritable, MapWritable> {
private IntWritable nKey = new IntWritable();
-
- public static void initJob(String path, Class<TransposeMap> map,
- Class<IntWritable> outputKeyClass, Class<MapWritable> outputValueClass,
- JobConf jobConf) {
- jobConf.setMapOutputValueClass(outputValueClass);
- jobConf.setMapOutputKeyClass(outputKeyClass);
- jobConf.setMapperClass(map);
+ public void map(ImmutableBytesWritable key, Result value, Context context)
+ throws IOException, InterruptedException {
+ IntWritable k = new IntWritable(BytesUtil.getRowIndex(key.get()));
- jobConf.setInputFormat(VectorInputFormat.class);
- FileInputFormat.addInputPaths(jobConf, path);
- jobConf.set(VectorInputFormat.COLUMN_LIST, Constants.COLUMN);
- }
-
- @Override
- public void map(IntWritable key, MapWritable value,
- OutputCollector<IntWritable, MapWritable> output, Reporter reporter)
- throws IOException {
-
- for(Map.Entry<Writable, Writable> e : value.entrySet()) {
+ for(Map.Entry<Writable, Writable> e : new DenseVector(value).getEntries().entrySet()) {
MapWritable val = new MapWritable();
nKey.set(((IntWritable) e.getKey()).get());
- val.put(key, e.getValue());
- output.collect(nKey, val);
+ val.put(k, e.getValue());
+ context.write(nKey, val);
val.clear();
}
Modified: incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeReduce.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeReduce.java?rev=828342&r1=828341&r2=828342&view=diff
==============================================================================
--- incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeReduce.java (original)
+++ incubator/hama/trunk/src/java/org/apache/hama/matrix/algebra/TransposeReduce.java Thu Oct 22 07:49:48 2009
@@ -20,47 +20,35 @@
package org.apache.hama.matrix.algebra;
import java.io.IOException;
-import java.util.Iterator;
import java.util.Map;
-import org.apache.hadoop.hbase.io.BatchUpdate;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reducer;
-import org.apache.hadoop.mapred.Reporter;
import org.apache.hama.io.VectorUpdate;
-import org.apache.hama.mapred.VectorOutputFormat;
+import org.apache.hama.util.BytesUtil;
-public class TransposeReduce extends MapReduceBase implements
-Reducer<IntWritable, MapWritable, IntWritable, VectorUpdate> {
-
- public static void initJob(String path, Class<TransposeReduce> reducer,
- JobConf jobConf) {
- jobConf.setOutputFormat(VectorOutputFormat.class);
- jobConf.setReducerClass(reducer);
- jobConf.set(VectorOutputFormat.OUTPUT_TABLE, path);
- jobConf.setOutputKeyClass(IntWritable.class);
- jobConf.setOutputValueClass(BatchUpdate.class);
- }
+public class TransposeReduce extends
+ TableReducer<IntWritable, MapWritable, Writable> {
@Override
- public void reduce(IntWritable key, Iterator<MapWritable> values,
- OutputCollector<IntWritable, VectorUpdate> output, Reporter reporter)
- throws IOException {
+ public void reduce(IntWritable key, Iterable<MapWritable> values,
+ Context context) throws IOException, InterruptedException {
MapWritable sum = new MapWritable();
- while (values.hasNext()) {
- for(Map.Entry<Writable, Writable> e: values.next().entrySet()) {
- sum.put(e.getKey(), e.getValue());
+ for (MapWritable value : values) {
+ for (Map.Entry<Writable, Writable> e : value.entrySet()) {
+ sum.put(e.getKey(), e.getValue());
}
}
-
+
VectorUpdate update = new VectorUpdate(key.get());
update.putAll(sum);
- output.collect(key, update);
+
+ context.write(new ImmutableBytesWritable(BytesUtil.getRowIndex(key.get())),
+ update.getPut());
}
-}
+
+}
\ No newline at end of file
Modified: incubator/hama/trunk/src/test/org/apache/hama/matrix/TestAbstractMatrix.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/src/test/org/apache/hama/matrix/TestAbstractMatrix.java?rev=828342&r1=828341&r2=828342&view=diff
==============================================================================
--- incubator/hama/trunk/src/test/org/apache/hama/matrix/TestAbstractMatrix.java (original)
+++ incubator/hama/trunk/src/test/org/apache/hama/matrix/TestAbstractMatrix.java Thu Oct 22 07:49:48 2009
@@ -34,7 +34,6 @@
public void testTransposeAndNorm() throws IOException {
testTrans(m1);
testTrans(m2);
-
normTest(m1);
normTest(m2);
}