TsvToAvroMapper extends Mapper<Writable, Text, AvroKey<DocumentSimilarity>, NullWritable> { @Override protected void map(Writable ignore, Text data, Context context) throws IOException, InterruptedException { try { String[] fields = data.toString().split("\\t"); DocumentSimilarity similarity = new DocumentSimilarity(); similarity.setDocumentId(fields[0]); similarity.setOtherDocumentId(fields[1]); similarity.setSimilarity(Float.parseFloat(fields[2])); context.write(new AvroKey<DocumentSimilarity>(similarity), NullWritable.get()); } catch (Exception e) { log.error(e); } }    }
@Test public void testMap() throws Exception { String docId = "doc1"; String otherDocId = "doc1"; float similarityLevel = 0.9f; mapper.map(null, new Text(buildLine(docId, otherDocId, similarityLevel)), context); verify(context).write(keyCaptor.capture(), valueCaptor.capture()); assertTrue(NullWritable.get() == valueCaptor.getValue()); DocumentSimilarity docSim = keyCaptor.getValue().datum(); assertNotNull(docSim); assertEquals(docId, docSim.getDocumentId()); assertEquals(otherDocId, docSim.getOtherDocumentId()); assertEquals(similarityLevel, docSim.getSimilarity().floatValue(), 0.00001); verifyZeroInteractions(log); }
@Test public void testMapForInvalidInput() throws Exception { String input = "invalid line"; mapper.map(null, new Text(input), context); verify(context, never()).write(any(), any()); verify(log).error(isA(ArrayIndexOutOfBoundsException.class)); }