MaybePruneRowsMapper extends Mapper<VarLongWritable,VectorWritable,IntWritable,DistributedRowMatrix.MatrixEntryWritable> { @Override protected void map(VarLongWritable rowIndex, VectorWritable vectorWritable, Context ctx) throws IOException, InterruptedException { Vector vector = vectorWritable.get(); countSeen(vector); int numElementsBeforePruning = vector.getNumNondefaultElements(); vector = maybePruneVector(vector); int numElementsAfterPruning = vector.getNumNondefaultElements(); ctx.getCounter(Elements.USED).increment(numElementsAfterPruning); ctx.getCounter(Elements.NEGLECTED).increment(numElementsBeforePruning - numElementsAfterPruning); DistributedRowMatrix.MatrixEntryWritable entry = new DistributedRowMatrix.MatrixEntryWritable(); int colIndex = TasteHadoopUtils.idToIndex(rowIndex.get()); entry.setCol(colIndex); Iterator<Vector.Element> iterator = vector.iterateNonZero(); while (iterator.hasNext()) { Vector.Element elem = iterator.next(); entry.setRow(elem.index()); entry.setVal(elem.get()); ctx.write(new IntWritable(elem.index()), entry); } }   static final String MAX_COOCCURRENCES; }
@Test public void testPruning() throws Exception { Vector v1 = new RandomAccessSparseVector(Integer.MAX_VALUE, 100); v1.set(1, 1); v1.set(3, 1); Vector v2 = new RandomAccessSparseVector(Integer.MAX_VALUE, 100); v2.set(1, 1); v2.set(7, 1); Vector v3 = new RandomAccessSparseVector(Integer.MAX_VALUE, 100); v3.set(1, 1); v3.set(5, 1); v3.set(9, 1); MaybePruneRowsMapper mapper = new MaybePruneRowsMapper(); setField(mapper, "maxCooccurrences", 2); Mapper<VarLongWritable,VectorWritable, IntWritable, DistributedRowMatrix.MatrixEntryWritable>.Context ctx = EasyMock.createMock(Mapper.Context.class); Counter usedElementsCounter = EasyMock.createMock(Counter.class); Counter neglectedElementsCounter = EasyMock.createMock(Counter.class); ctx.write(EasyMock.eq(new IntWritable(1)), MathHelper.matrixEntryMatches(1, 123, 1)); ctx.write(EasyMock.eq(new IntWritable(3)), MathHelper.matrixEntryMatches(3, 123, 1)); EasyMock.expect(ctx.getCounter(MaybePruneRowsMapper.Elements.USED)).andReturn(usedElementsCounter); usedElementsCounter.increment(2); EasyMock.expect(ctx.getCounter(MaybePruneRowsMapper.Elements.NEGLECTED)).andReturn(neglectedElementsCounter); neglectedElementsCounter.increment(0); ctx.write(EasyMock.eq(new IntWritable(1)), MathHelper.matrixEntryMatches(1, 456, 1)); ctx.write(EasyMock.eq(new IntWritable(7)), MathHelper.matrixEntryMatches(7, 456, 1)); EasyMock.expect(ctx.getCounter(MaybePruneRowsMapper.Elements.USED)).andReturn(usedElementsCounter); usedElementsCounter.increment(2); EasyMock.expect(ctx.getCounter(MaybePruneRowsMapper.Elements.NEGLECTED)).andReturn(neglectedElementsCounter); neglectedElementsCounter.increment(0); ctx.write(EasyMock.eq(new IntWritable(5)), MathHelper.matrixEntryMatches(5, 789, 1)); ctx.write(EasyMock.eq(new IntWritable(9)), MathHelper.matrixEntryMatches(9, 789, 1)); EasyMock.expect(ctx.getCounter(MaybePruneRowsMapper.Elements.USED)).andReturn(usedElementsCounter); usedElementsCounter.increment(2); EasyMock.expect(ctx.getCounter(MaybePruneRowsMapper.Elements.NEGLECTED)).andReturn(neglectedElementsCounter); neglectedElementsCounter.increment(1); EasyMock.replay(ctx, usedElementsCounter, neglectedElementsCounter); mapper.map(new VarLongWritable(123L), new VectorWritable(v1), ctx); mapper.map(new VarLongWritable(456L), new VectorWritable(v2), ctx); mapper.map(new VarLongWritable(789L), new VectorWritable(v3), ctx); EasyMock.verify(ctx, usedElementsCounter, neglectedElementsCounter); }