@Test(dependsOnMethods = "testSerializeToSequenceFile") public void testDeserializeFromSequenceFile() throws IOException { Queue<WorkUnitState> workUnitStates = Queues.newConcurrentLinkedQueue(); Path seqPath1 = new Path(this.outputPath, "seq1"); Path seqPath2 = new Path(this.outputPath, "seq2"); try (ParallelRunner parallelRunner = new ParallelRunner(2, this.fs)) { parallelRunner.deserializeFromSequenceFile(Text.class, WorkUnitState.class, seqPath1, workUnitStates, true); parallelRunner.deserializeFromSequenceFile(Text.class, WorkUnitState.class, seqPath2, workUnitStates, true); } Assert.assertFalse(this.fs.exists(seqPath1)); Assert.assertFalse(this.fs.exists(seqPath2)); Assert.assertEquals(workUnitStates.size(), 2); for (WorkUnitState workUnitState : workUnitStates) { TestWatermark watermark = new Gson().fromJson(workUnitState.getActualHighWatermark(), TestWatermark.class); Assert.assertTrue(watermark.getLongWatermark() == 10L || watermark.getLongWatermark() == 100L); } }
public <T extends State> void deserializeFromSequenceFile(final Class<? extends Writable> keyClass, final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter) { this.futures.add(new NamedFuture(this.executor.submit(new Callable<Void>() { @Override public Void call() throws Exception { Configuration conf = new Configuration(ParallelRunner.this.fs.getConf()); WritableShimSerialization.addToHadoopConfiguration(conf); try (@SuppressWarnings("deprecation") SequenceFile.Reader reader = new SequenceFile.Reader( ParallelRunner.this.fs, inputFilePath, conf)) { Writable key = keyClass.newInstance(); T state = stateClass.newInstance(); while (reader.next(key)) { state = (T) reader.getCurrentValue(state); states.add(state); state = stateClass.newInstance(); } if (deleteAfter) { HadoopUtils.deletePath(ParallelRunner.this.fs, inputFilePath, false); } } return null; } }), "Deserialize state from file " + inputFilePath)); }
ParallelRunner implements Closeable { public <T extends State> void deserializeFromSequenceFile(final Class<? extends Writable> keyClass, final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter) { this.futures.add(new NamedFuture(this.executor.submit(new Callable<Void>() { @Override public Void call() throws Exception { Configuration conf = new Configuration(ParallelRunner.this.fs.getConf()); WritableShimSerialization.addToHadoopConfiguration(conf); try (@SuppressWarnings("deprecation") SequenceFile.Reader reader = new SequenceFile.Reader( ParallelRunner.this.fs, inputFilePath, conf)) { Writable key = keyClass.newInstance(); T state = stateClass.newInstance(); while (reader.next(key)) { state = (T) reader.getCurrentValue(state); states.add(state); state = stateClass.newInstance(); } if (deleteAfter) { HadoopUtils.deletePath(ParallelRunner.this.fs, inputFilePath, false); } } return null; } }), "Deserialize state from file " + inputFilePath)); } }
ParallelRunner implements Closeable { public <T extends State> void deserializeFromSequenceFile(final Class<? extends Writable> keyClass, final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter) { this.futures.add(new NamedFuture(this.executor.submit(new Callable<Void>() { @Override public Void call() throws Exception { Configuration conf = new Configuration(ParallelRunner.this.fs.getConf()); WritableShimSerialization.addToHadoopConfiguration(conf); try (@SuppressWarnings("deprecation") SequenceFile.Reader reader = new SequenceFile.Reader( ParallelRunner.this.fs, inputFilePath, conf)) { Writable key = keyClass.newInstance(); T state = stateClass.newInstance(); while (reader.next(key)) { state = (T) reader.getCurrentValue(state); states.add(state); state = stateClass.newInstance(); } if (deleteAfter) { HadoopUtils.deletePath(ParallelRunner.this.fs, inputFilePath, false); } } return null; } }), "Deserialize state from file " + inputFilePath)); } ParallelRunner(int threads, FileSystem fs);  ParallelRunner(int threads, FileSystem fs, FailPolicy failPolicy); }
ParallelRunner implements Closeable { public <T extends State> void deserializeFromSequenceFile(final Class<? extends Writable> keyClass, final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter) { this.futures.add(new NamedFuture(this.executor.submit(new Callable<Void>() { @Override public Void call() throws Exception { Configuration conf = new Configuration(ParallelRunner.this.fs.getConf()); WritableShimSerialization.addToHadoopConfiguration(conf); try (@SuppressWarnings("deprecation") SequenceFile.Reader reader = new SequenceFile.Reader( ParallelRunner.this.fs, inputFilePath, conf)) { Writable key = keyClass.newInstance(); T state = stateClass.newInstance(); while (reader.next(key)) { state = (T) reader.getCurrentValue(state); states.add(state); state = stateClass.newInstance(); } if (deleteAfter) { HadoopUtils.deletePath(ParallelRunner.this.fs, inputFilePath, false); } } return null; } }), "Deserialize state from file " + inputFilePath)); } ParallelRunner(int threads, FileSystem fs);  ParallelRunner(int threads, FileSystem fs, FailPolicy failPolicy); void serializeToFile(final T state, final Path outputFilePath); void deserializeFromFile(final T state, final Path inputFilePath); void deserializeFromSequenceFile(final Class<? extends Writable> keyClass,
      final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter); void deletePath(final Path path, final boolean recursive); void renamePath(final Path src, final Path dst, final Optional<String> group); void movePath(final Path src, final FileSystem dstFs, final Path dst, final Optional<String> group); void movePath(final Path src, final FileSystem dstFs, final Path dst, final boolean overwrite,
      final Optional<String> group); void submitCallable(Callable<Void> callable, String name); void waitForTasks(long timeoutInMills); void waitForTasks(); @Override void close(); }
ParallelRunner implements Closeable { public <T extends State> void deserializeFromSequenceFile(final Class<? extends Writable> keyClass, final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter) { this.futures.add(new NamedFuture(this.executor.submit(new Callable<Void>() { @Override public Void call() throws Exception { Configuration conf = new Configuration(ParallelRunner.this.fs.getConf()); WritableShimSerialization.addToHadoopConfiguration(conf); try (@SuppressWarnings("deprecation") SequenceFile.Reader reader = new SequenceFile.Reader( ParallelRunner.this.fs, inputFilePath, conf)) { Writable key = keyClass.newInstance(); T state = stateClass.newInstance(); while (reader.next(key)) { state = (T) reader.getCurrentValue(state); states.add(state); state = stateClass.newInstance(); } if (deleteAfter) { HadoopUtils.deletePath(ParallelRunner.this.fs, inputFilePath, false); } } return null; } }), "Deserialize state from file " + inputFilePath)); } ParallelRunner(int threads, FileSystem fs);  ParallelRunner(int threads, FileSystem fs, FailPolicy failPolicy); void serializeToFile(final T state, final Path outputFilePath); void deserializeFromFile(final T state, final Path inputFilePath); void deserializeFromSequenceFile(final Class<? extends Writable> keyClass,
      final Class<T> stateClass, final Path inputFilePath, final Collection<T> states, final boolean deleteAfter); void deletePath(final Path path, final boolean recursive); void renamePath(final Path src, final Path dst, final Optional<String> group); void movePath(final Path src, final FileSystem dstFs, final Path dst, final Optional<String> group); void movePath(final Path src, final FileSystem dstFs, final Path dst, final boolean overwrite,
      final Optional<String> group); void submitCallable(Callable<Void> callable, String name); void waitForTasks(long timeoutInMills); void waitForTasks(); @Override void close(); static final String PARALLEL_RUNNER_THREADS_KEY; static final int DEFAULT_PARALLEL_RUNNER_THREADS; }