AccumuloAddElementsFromHdfsJobFactory implements AddElementsFromHdfsJobFactory { @Override public void setupJob(final Job job, final AddElementsFromHdfs operation, final String mapperGenerator, final Store store) throws IOException { job.setJarByClass(getClass()); job.setJobName(getJobName(mapperGenerator, operation.getOutputPath())); setupMapper(job); setupCombiner(job); setupReducer(job); setupOutput(job, operation); if (!NoPartitioner.class.equals(operation.getPartitioner())) { if (null != operation.getPartitioner()) { operation.setPartitioner(GafferKeyRangePartitioner.class); LOGGER.warn("Partitioner class " + operation.getPartitioner().getName() + " will be replaced with " + GafferKeyRangePartitioner.class.getName()); } setupPartitioner(job, operation, (AccumuloStore) store); } } AccumuloAddElementsFromHdfsJobFactory(final Configuration configuration);  AccumuloAddElementsFromHdfsJobFactory(); @Override void prepareStore(final Store store); @Override JobConf createJobConf(final AddElementsFromHdfs operation, final String mapperGeneratorClassName, final Store store); @Override void setupJob(final Job job, final AddElementsFromHdfs operation, final String mapperGenerator, final Store store); static final String INVALID_FIELD_MUST_BE_GREATER_THAN_OR_EQUAL_TO_ONE_GOT_S; static final String INGEST_HDFS_DATA_GENERATOR_S_OUTPUT_S; }
@Test public void shouldSetupJob() throws IOException { final JobConf localConf = createLocalConf(); final FileSystem fs = FileSystem.getLocal(localConf); fs.mkdirs(new Path(outputDir)); fs.mkdirs(new Path(splitsDir)); try (final BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fs.create(new Path(splitsFile), true)))) { writer.write("1"); } final AccumuloAddElementsFromHdfsJobFactory factory = getJobFactory(); final Job job = mock(Job.class); final AddElementsFromHdfs operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir.toString()) .addInputMapperPair(inputDir.toString(), TextMapperGeneratorImpl.class.getName()) .useProvidedSplits(true) .splitsFilePath(splitsFile.toString()) .build(); final AccumuloStore store = mock(AccumuloStore.class); given(job.getConfiguration()).willReturn(localConf); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); verify(job).setJarByClass(factory.getClass()); verify(job).setJobName(String.format(AccumuloAddElementsFromHdfsJobFactory.INGEST_HDFS_DATA_GENERATOR_S_OUTPUT_S, TextMapperGeneratorImpl.class.getName(), outputDir)); verify(job).setMapperClass(AddElementsFromHdfsMapper.class); verify(job).setMapOutputKeyClass(Key.class); verify(job).setMapOutputValueClass(Value.class); verify(job).setCombinerClass(AccumuloKeyValueReducer.class); verify(job).setReducerClass(AccumuloKeyValueReducer.class); verify(job).setOutputKeyClass(Key.class); verify(job).setOutputValueClass(Value.class); job.setOutputFormatClass(AccumuloFileOutputFormat.class); assertEquals(fs.makeQualified(new Path(outputDir)).toString(), job.getConfiguration().get("mapreduce.output.fileoutputformat.outputdir")); verify(job).setNumReduceTasks(2); verify(job).setPartitionerClass(GafferKeyRangePartitioner.class); assertEquals(splitsFile, job.getConfiguration().get(GafferRangePartitioner.class.getName() + ".cutFile")); }
@Test public void shouldSetNoMoreThanMaxNumberOfReducersSpecified() throws IOException, StoreException, OperationException { store.initialise("graphId", SCHEMA, PROPERTIES); final JobConf localConf = createLocalConf(); final FileSystem fs = FileSystem.getLocal(localConf); fs.mkdirs(new Path(outputDir)); fs.mkdirs(new Path(splitsDir)); final BufferedWriter writer = new BufferedWriter(new FileWriter(splitsFile.toString())); for (int i = 100; i < 200; i++) { writer.write(i + "\n"); } writer.close(); final SplitStoreFromFile splitTable = new SplitStoreFromFile.Builder() .inputPath(splitsFile.toString()) .build(); store.execute(splitTable, new Context(new User())); final AccumuloAddElementsFromHdfsJobFactory factory = getJobFactory(); final Job job = Job.getInstance(localConf); AddElementsFromHdfs operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir.toString()) .addInputMapperPair(inputDir.toString(), TextMapperGeneratorImpl.class.getName()) .maxReducers(10) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() <= 10); operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir.toString()) .addInputMapperPair(inputDir.toString(), TextMapperGeneratorImpl.class.getName()) .maxReducers(100) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() <= 100); operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir.toString()) .addInputMapperPair(inputDir.toString(), TextMapperGeneratorImpl.class.getName()) .maxReducers(1000) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() <= 1000); }
@Test public void shouldSetNoLessThanMinNumberOfReducersSpecified() throws IOException, StoreException, OperationException { store.initialise("graphId", SCHEMA, PROPERTIES); final JobConf localConf = createLocalConf(); final FileSystem fs = FileSystem.getLocal(localConf); fs.mkdirs(new Path(outputDir)); fs.mkdirs(new Path(splitsDir)); final BufferedWriter writer = new BufferedWriter(new FileWriter(splitsFile)); for (int i = 100; i < 200; i++) { writer.write(i + "\n"); } writer.close(); final SplitStoreFromFile splitTable = new SplitStoreFromFile.Builder() .inputPath(splitsFile) .build(); store.execute(splitTable, new Context(new User())); final AccumuloAddElementsFromHdfsJobFactory factory = getJobFactory(); final Job job = Job.getInstance(localConf); AddElementsFromHdfs operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(10) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() >= 10); operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(100) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() >= 100); operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(1000) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() >= 1000); }
@Test public void shouldSetNumberOfReducersBetweenMinAndMaxSpecified() throws IOException, StoreException, OperationException { store.initialise("graphId", SCHEMA, PROPERTIES); final JobConf localConf = createLocalConf(); final FileSystem fs = FileSystem.getLocal(localConf); fs.mkdirs(new Path(outputDir)); fs.mkdirs(new Path(splitsDir)); final BufferedWriter writer = new BufferedWriter(new FileWriter(splitsFile)); for (int i = 100; i < 200; i++) { writer.write(i + "\n"); } writer.close(); final SplitStoreFromFile splitTable = new SplitStoreFromFile.Builder() .inputPath(splitsFile) .build(); store.execute(splitTable, new Context(new User())); final AccumuloAddElementsFromHdfsJobFactory factory = getJobFactory(); final Job job = Job.getInstance(localConf); AddElementsFromHdfs operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(10) .maxReducers(20) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() >= 10); assertTrue(job.getNumReduceTasks() <= 20); operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(100) .maxReducers(200) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() >= 100); assertTrue(job.getNumReduceTasks() <= 200); operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(1000) .maxReducers(2000) .splitsFilePath("target/data/splits.txt") .build(); factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); assertTrue(job.getNumReduceTasks() >= 1000); assertTrue(job.getNumReduceTasks() <= 2000); }
@Test public void shouldThrowExceptionWhenMaxReducersSetOutsideOfRange() throws IOException, StoreException, OperationException { store.initialise("graphId", SCHEMA, PROPERTIES); final JobConf localConf = createLocalConf(); final FileSystem fs = FileSystem.getLocal(localConf); fs.mkdirs(new Path(outputDir)); fs.mkdirs(new Path(splitsDir)); final BufferedWriter writer = new BufferedWriter(new FileWriter(splitsFile)); for (int i = 100; i < 200; i++) { writer.write(i + "\n"); } writer.close(); final SplitStoreFromFile splitTable = new SplitStoreFromFile.Builder() .inputPath(splitsFile) .build(); store.execute(splitTable, new Context(new User())); final AccumuloAddElementsFromHdfsJobFactory factory = getJobFactory(); final Job job = Job.getInstance(localConf); AddElementsFromHdfs operation = new AddElementsFromHdfs.Builder() .outputPath(outputDir) .addInputMapperPair(inputDir, TextMapperGeneratorImpl.class.getName()) .minReducers(100) .maxReducers(101) .splitsFilePath("target/data/splits.txt") .build(); try { factory.setupJob(job, operation, TextMapperGeneratorImpl.class.getName(), store); fail("Exception expected"); } catch (IllegalArgumentException e) { assertTrue(e.getMessage().contains("not a valid range")); } }