@Test public void testLineRecordReader() throws Exception { File dataFile = new ClassPathResource("iris.dat").getFile(); List<String> lines = FileUtils.readLines(dataFile); List<String> linesRdd = (lines); CSVRecordReader rr = new CSVRecordReader(0, ','); List<List<Writable>> out = linesRdd.stream().map(input -> new LineRecordReaderFunction(rr).apply(input)).collect(Collectors.toList()); List<List<Writable>> outList = out; CSVRecordReader rr2 = new CSVRecordReader(0, ','); rr2.initialize(new FileSplit(dataFile)); Set<List<Writable>> expectedSet = new HashSet<>(); int totalCount = 0; while (rr2.hasNext()) { expectedSet.add(rr2.next()); totalCount++; } assertEquals(totalCount, outList.size()); for (List<Writable> line : outList) { assertTrue(expectedSet.contains(line)); } }
@Override public List<Writable> apply(String s) { try { recordReader.initialize(new StringSplit(s)); } catch (Exception e) { throw new IllegalStateException(e); } return recordReader.next(); }
LineRecordReaderFunction implements Function<String, List<Writable>> { @Override public List<Writable> apply(String s) { try { recordReader.initialize(new StringSplit(s)); } catch (Exception e) { throw new IllegalStateException(e); } return recordReader.next(); } }
LineRecordReaderFunction implements Function<String, List<Writable>> { @Override public List<Writable> apply(String s) { try { recordReader.initialize(new StringSplit(s)); } catch (Exception e) { throw new IllegalStateException(e); } return recordReader.next(); } LineRecordReaderFunction(RecordReader recordReader); }
LineRecordReaderFunction implements Function<String, List<Writable>> { @Override public List<Writable> apply(String s) { try { recordReader.initialize(new StringSplit(s)); } catch (Exception e) { throw new IllegalStateException(e); } return recordReader.next(); } LineRecordReaderFunction(RecordReader recordReader); @Override List<Writable> apply(String s); }
LineRecordReaderFunction implements Function<String, List<Writable>> { @Override public List<Writable> apply(String s) { try { recordReader.initialize(new StringSplit(s)); } catch (Exception e) { throw new IllegalStateException(e); } return recordReader.next(); } LineRecordReaderFunction(RecordReader recordReader); @Override List<Writable> apply(String s);  }