@Test public void testCsvSimple() throws Exception { File dir = testDir.newFolder(); File f1 = new File(dir, "file1.txt"); File f2 = new File(dir, "file2.txt"); FileUtils.writeStringToFile(f1, "a,b,c\nd,e,f", StandardCharsets.UTF_8); FileUtils.writeStringToFile(f2, "1,2,3", StandardCharsets.UTF_8); List<URI> uris = Arrays.asList(f1.toURI(), f2.toURI()); CSVRecordReader rr = new CSVRecordReader(); TestStreamFunction fn = new TestStreamFunction(); InputSplit is = new StreamInputSplit(uris, fn); rr.initialize(is); List<List<Writable>> exp = new ArrayList<>(); exp.add(Arrays.<Writable>asList(new Text("a"), new Text("b"), new Text("c"))); exp.add(Arrays.<Writable>asList(new Text("d"), new Text("e"), new Text("f"))); exp.add(Arrays.<Writable>asList(new Text("1"), new Text("2"), new Text("3"))); List<List<Writable>> act = new ArrayList<>(); while(rr.hasNext()){ act.add(rr.next()); } assertEquals(exp, act); assertEquals(uris, fn.calledWithUris); rr.reset(); int count = 0; while(rr.hasNext()) { count++; rr.next(); } assertEquals(3, count); }
@Override public void reset() { }
StreamInputSplit implements InputSplit { @Override public void reset() { } }
StreamInputSplit implements InputSplit { @Override public void reset() { } StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn);  StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn, Random rng); }
StreamInputSplit implements InputSplit { @Override public void reset() { } StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn);  StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn, Random rng); @Override boolean canWriteToLocation(URI location); @Override String addNewLocation(); @Override String addNewLocation(String location); @Override void updateSplitLocations(boolean reset); @Override boolean needsBootstrapForWrite(); @Override void bootStrapForWrite(); @Override OutputStream openOutputStreamFor(String location); @Override InputStream openInputStreamFor(String location); @Override long length(); @Override URI[] locations(); @Override Iterator<URI> locationsIterator(); @Override Iterator<String> locationsPathIterator(); @Override void reset(); @Override boolean resetSupported(); }
StreamInputSplit implements InputSplit { @Override public void reset() { } StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn);  StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn, Random rng); @Override boolean canWriteToLocation(URI location); @Override String addNewLocation(); @Override String addNewLocation(String location); @Override void updateSplitLocations(boolean reset); @Override boolean needsBootstrapForWrite(); @Override void bootStrapForWrite(); @Override OutputStream openOutputStreamFor(String location); @Override InputStream openInputStreamFor(String location); @Override long length(); @Override URI[] locations(); @Override Iterator<URI> locationsIterator(); @Override Iterator<String> locationsPathIterator(); @Override void reset(); @Override boolean resetSupported();  }