IngestJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption(JOB_NAME_OPTION, "jn", "Optional name to give the Hadoop Job.", false); addOption(ZK_CONNECT_OPTION, "zk", "Connection string for ZooKeeper. Either Solr Server or ZK must be specified.", false); addOption(SOLR_SERVER_OPTION, "s", "The URL for Solr, when not in SolrCloud. Either Solr Server or ZK must be specified.", false); addOption(COLLECTION_OPTION, "c", "Collection name", true); addOption(MIME_TYPE_OPTION, "mt", "a mimeType known to Tika", false); addOption(CONF_OPTION, "con", "Key[value] pairs, separated by ; to be added to the Conf. The value is set as a string.", false); addOption(MAPPER_OPTION, "cls", "Fully qualified class name for the IngestMapper to use (derived from AbstractIngestMapper)", false); addOption(REDUCER_OPTION, "redcls", "Fully qualified class name for your custom IngestReducer. You must also set " + NUM_REDUCERS_OPTION + " > 0 for this to be invoked.", false); addOption(OVERWRITE_OPTION, "ov", "Overwrite any existing documents", "true"); addOption(INPUT_FORMAT, "if", "The Hadoop InputFormat type to use. Must be a fully qualified class name and the class must be available on the classpath. " + " Note: Some Mapper Classes may ignore this. Also note, your Input Format must be able to produce the correct kind of records for the mapper."); addOption(OUTPUT_FORMAT, "of", "The Hadoop OutputFormat type to use. Must be a fully qualified class name and the class must be available on the classpath. " + " Note: Some Mapper Classes may ignore this. Also note, your Output Format must be able to handle <Text, DocumentWritable>."); addOption(NUM_REDUCERS_OPTION, "ur", "An Integer >= 0 indicating the number of Reducers to use when outputing to the OutputFormat. " + "Depending on the OutputFormat and your system resources, you may wish to have Hadoop do a reduce step first so as to not open too many connections to the output resource. Default is to not use any reducers. Note, "); addFlag(QUIET, "q", "Quiet option for common.AbstractJob"); Map<String, List<String>> map = parseArguments(args, false, true); if (map == null) { return 1; } Path input = getInputPath(); Path output = getOutputPath(); JobConf conf = new JobConf(getConf(), getClass()); conf.setJarByClass(IngestJob.class); conf.setJobName(getOption(JOB_NAME_OPTION, IngestJob.class.getName())); String zk = getOption(ZK_CONNECT_OPTION); String solr = getOption(SOLR_SERVER_OPTION); conf.set(COLLECTION, getOption(COLLECTION_OPTION)); conf.set(LucidWorksWriter.SOLR_COLLECTION, getOption(COLLECTION_OPTION)); conf.set(MIME_TYPE, getOption(MIME_TYPE_OPTION, "")); conf.set(TEMP_DIR, getTempPath().toString()); boolean overwrite = Boolean.parseBoolean(getOption(OVERWRITE_OPTION, "true")); conf.set(OVERWRITE, overwrite ? "true" : "false"); String confAdds = getOption(CONF_OPTION); if (confAdds != null && !confAdds.isEmpty()) { processConf(confAdds, conf); } conf.setMapOutputKeyClass(Text.class); conf.setMapOutputValueClass(LWDocumentWritable.class); conf.setOutputKeyClass(Text.class); conf.setOutputValueClass(LWDocumentWritable.class); int numReducers = Integer.parseInt(getOption(NUM_REDUCERS_OPTION, "0")); conf.setNumReduceTasks(numReducers); conf.set("mapred.map.tasks.speculative.execution", "false"); conf.set("mapred.reduce.tasks.speculative.execution", "false"); conf.set("mapreduce.reduce.speculative", "false"); conf.set("mapreduce.map.speculative", "false"); FileInputFormat.setInputPaths(conf, input); if (output != null) { FileOutputFormat.setOutputPath(conf, output); } String mapperClass = getOption(MAPPER_OPTION, DEFAULT_MAPPER_CLASS); Class<AbstractIngestMapper> mapperClazz; AbstractIngestMapper mapper; try { mapperClazz = (Class<AbstractIngestMapper>) Class.forName(mapperClass); mapper = mapperClazz.newInstance(); } catch (Exception e) { System.out.println("Unable to instantiate AbstractIngestMapper class"); log.error("Unable to instantiate AbstractIngestMapper class " + mapperClass, e); return 1; } conf.setMapperClass(mapperClazz); String inputFormatName = getOption(INPUT_FORMAT); if (inputFormatName != null) { Class<? extends InputFormat> ifClass = Class.forName(inputFormatName) .asSubclass(InputFormat.class); conf.setInputFormat(ifClass); conf.setBoolean(INPUT_FORMAT_OVERRIDE, true); } String reducerClass = getOption(REDUCER_OPTION, IngestReducer.class.getName()); IngestReducer reducer = null; if (numReducers > 0 && reducerClass != null) { Class<IngestReducer> reducerClazz; try { reducerClazz = (Class<IngestReducer>) Class.forName(reducerClass); reducer = reducerClazz.newInstance(); } catch (Exception e) { System.out.println("Unable to instantiate IngestReducer class"); log.error("Unable to instantiate IngestReducer class " + reducerClass, e); return 1; } conf.setReducerClass(reducerClazz); } String outputFormatName = getOption(OUTPUT_FORMAT); if (outputFormatName == null || outputFormatName.equals(LWMapRedOutputFormat.class.getName())) { log.info("Using default OutputFormat: {}", LWMapRedOutputFormat.class.getName()); conf.setOutputFormat(LWMapRedOutputFormat.class); if (!checkSolrOrZkString(conf)) { System.out.println("You must specify either the " + ZK_CONNECT_OPTION + " or the " + SOLR_SERVER_OPTION); return 1; } if (!checkSolrServer(conf)) { System.out.println("Solr server not available on: " + (zk == null || zk.isEmpty() ? solr : zk)); System.out.println("Make sure that collection [" + getOption(COLLECTION_OPTION) + "] exists"); return 1; } } else { log.info("OutputFormat is {}", outputFormatName); Class<? extends OutputFormat> ofClass = Class.forName(outputFormatName).asSubclass(OutputFormat.class); conf.setOutputFormat(ofClass); conf.setBoolean(OUTPUT_FORMAT_OVERRIDE, true); } mapper.getFixture().init(conf); if (reducer != null) { reducer.getFixture().init(conf); } boolean debugAll = conf.getBoolean("lww.debug.all", false); if (debugAll) { conf.set("yarn.app.mapreduce.am.log.level", "DEBUG"); } RunningJob job = JobClient.runJob(conf); job.waitForCompletion(); mapper.getFixture().close(); if (reducer != null) { reducer.getFixture().close(); } doFinalCommit(conf, job); if (job.isSuccessful()) { long added = job.getCounters().getCounter(BaseHadoopIngest.Counters.DOCS_ADDED); if (added == 0) { System.out.println("Didn't ingest any documents, failing"); log.warn("Didn't ingest any documents, failing"); return 1; } else { return 0; } } else { log.warn("Something went wrong, failing"); return 1; } }  static void main(String[] args); @Override int run(String[] args); void doFinalCommit(JobConf conf, RunningJob job); boolean checkSolrServer(JobConf conf); static final String COLLECTION_OPTION; static final String MIME_TYPE_OPTION; static final String ZK_CONNECT_OPTION; static final String MAPPER_OPTION; static final String REDUCER_OPTION; static final String OVERWRITE_OPTION; static final String INPUT_FORMAT; static final String SOLR_SERVER_OPTION; static final String NUM_REDUCERS_OPTION; static final String OUTPUT_FORMAT; static final String DEFAULT_MAPPER_CLASS; static final String INPUT_FORMAT_OVERRIDE; static final String OUTPUT_FORMAT_OVERRIDE; static final String CONF_OPTION; static final String JOB_NAME_OPTION; static final String QUIET; }
@Test public void testGrok() throws Exception { String grok = "grok" + File.separator + "ip-word.log"; File grokFile = new File(ClassLoader.getSystemClassLoader().getResource(grok).getPath()); assertTrue(grok + " does not exist: " + grokFile.getAbsolutePath(), grokFile.exists()); Path input = new Path(tempDir, grok); addContentToFS(input, Files.toByteArray(grokFile)); String grokConf = "grok" + File.separator + "IP-WORD.conf"; File grokConfFile = new File(ClassLoader.getSystemClassLoader().getResource(grokConf).getPath()); assertTrue(grokConf + " does not exist: " + grokConfFile.getAbsolutePath(), grokConfFile.exists()); String jobName = "testGrok"; String[] args = new JobArgs().withJobName(jobName).withClassname(GrokIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).withDArgs("-Dgrok.uri=" + grokConfFile) .getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); MockRecordWriter mockRecordWriter = IngestJobMockMapRedOutFormat.writers.get(jobName); Assert.assertNotNull(mockRecordWriter); assertEquals(4000, mockRecordWriter.map.size()); }
@Test public void testReducer() throws Exception { Path input = new Path(tempDir, "reducer.csv"); StringBuilder buffer = new StringBuilder("id,bar,junk,zen,hockey"); buffer.append(lineSep).append("id-1, The quick brown fox, jumped, " + "head, gretzky, extra").append(lineSep) .append("id-2, The quick red fox, kicked, head," + "" + " gretzky"); addContentToFS(input, buffer.toString()); String jobName = "testCsvReducers"; String[] args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).withReducersClass(IngestReducer.class.getName()) .withReducersAmount("3") .withConf("csvFieldMapping[0=id," + "1=bar, 2=junk , 3 = zen ,4 = hockey];" + "idField[id];csvFirstLineComment[true]") .getJobArgs(); conf.set("io.serializations", "com.lucidworks.hadoop.io.impl.LWMockSerealization"); conf.set("io.sort.mb", "1"); ToolRunner.run(conf, new IngestJob(), args); verifyJob(jobName, 2, null, "hockey", "field_5"); }
@Test public void testBadArgs() throws Exception { String jobName = "testDidnotIngetAnyDocs"; String[] args = new JobArgs().withJobName(jobName).withClassname(DirectoryIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(tempDir.toUri().toString()).getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(1, val); assertErrorMessage("Didn't ingest any document"); Path input = new Path(tempDir, "foo.csv"); StringBuilder buffer = new StringBuilder("id,bar,junk,zen,hockey"); buffer.append(lineSep).append("id-1, The quick brown fox, jumped, " + "head, gretzky, extra").append(lineSep) .append("id-2, The quick red fox, kicked, head," + " gretzky"); addContentToFS(input, buffer.toString()); jobName = "testBadMapper"; args = new JobArgs().withJobName(jobName).withClassname("foo").withCollection(DEFAULT_COLLECTION) .withZkString(getBaseUrl()).withInput(input.toUri().toString()).getJobArgs(); val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(1, val); assertErrorMessage("Unable to instantiate AbstractIngestMapper class"); jobName = "testInvalidSolrConnection"; String invalidSolrConnection = getBaseUrl() + "+1"; args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(invalidSolrConnection) .withInput(input.toUri().toString()).withOutputFormat(LWMapRedOutputFormat.class.getName()) .getJobArgs(); val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(1, val); assertErrorMessage("server not available on"); jobName = "testBadReducer"; args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).withReducersClass("foo").withReducersAmount("3") .getJobArgs(); val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(1, val); assertErrorMessage("Unable to instantiate IngestReducer class"); jobName = "testNoZKorS"; args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withInput(input.toUri().toString()) .withOutputFormat(LWMapRedOutputFormat.class.getName()).getJobArgs(); val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(1, val); assertErrorMessage("You must specify either the.*or the"); jobName = "testNullArgs"; val = ToolRunner.run(conf, new IngestJob(), null); assertEquals(1, val); assertErrorMessage("Missing required option "); }
@Test public void testPingWrongCollection() throws Exception { String jobName = "testInvalidSolrConnection"; String invalidSolrConnection = getBaseUrl(); Path input = new Path(tempDir, "foo.csv"); StringBuilder buffer = new StringBuilder("id,bar,junk,zen,hockey"); buffer.append(lineSep).append("id-1, The quick brown fox, jumped, " + "head, gretzky, extra").append(lineSep) .append("id-2, The quick red fox, kicked, head," + "" + " gretzky"); addContentToFS(input, buffer.toString()); String[] args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection("INVALID-COLLECTION").withZkString(invalidSolrConnection) .withInput(input.toUri().toString()) .withConf("csvFieldMapping[0=id,1=bar, 2=junk " + ", 3 = zen ,4 = hockey];idField[id]") .withOutputFormat(LWMapRedOutputFormat.class.getName()).getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(1, val); assertErrorMessage("Make sure that collection"); }
@Test public void testXML() throws Exception { String xsl = "xml" + File.separator + "xml_ingest_mapper.xsl"; File xslFile = new File(ClassLoader.getSystemClassLoader().getResource(xsl).getPath()); assertTrue(xsl + " does not exist: " + xslFile.getAbsolutePath(), xslFile.exists()); Path inputXsl = new Path(tempDir, xsl); addContentToFS(inputXsl, Files.toByteArray(xslFile)); String xml = "xml" + File.separator + "foo.xml"; File xmlFile = new File(ClassLoader.getSystemClassLoader().getResource(xml).getPath()); assertTrue(xml + " does not exist: " + xmlFile.getAbsolutePath(), xmlFile.exists()); Path input = new Path(tempDir, xml); addContentToFS(input, Files.toByteArray(xmlFile)); String jobName = "testXml"; String[] args = new JobArgs().withJobName(jobName).withClassname(XMLIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).withConf("lww.xslt[" + inputXsl + "];lww.xml" + ".start[root]; lww.xml.end[root];lww.xml.docXPathExpr[ .getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 2, new String[]{"1", "2"}, "text", "int"); }
@Test public void testCSV() throws Exception { Path input = new Path(tempDir, "foo.csv"); StringBuilder buffer = new StringBuilder("id,bar,junk,zen,hockey"); buffer.append(lineSep).append("id-1, The quick brown fox, jumped, " + "head, gretzky, extra").append(lineSep) .append("id-2, The quick red fox, kicked, head," + " gretzky"); addContentToFS(input, buffer.toString()); String jobName = "testCsv"; String[] args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()) .withConf("csvFieldMapping[0=id,1=bar, 2=junk , 3 = zen ,4 = hockey];idField[id];" + "csvFirstLineComment[true]") .getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 2, null); jobName = "testCsv2"; args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()) .withConf("csvFieldMapping[0=id,1=bar, 2=junk , 3 = zen ,4 = hockey];idField[id]").getJobArgs(); ToolRunner.run(conf, new IngestJob(), args); verifyJob(jobName, 3, null); jobName = "testCsvFieldId"; args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()) .withConf("csvFieldMapping[0=bar, 1=id, 2=junk , 3 = zen ,4 = hockey];idField[junk]") .getJobArgs(); ToolRunner.run(conf, new IngestJob(), args); verifyJob(jobName, 3, null); }
@Test public void testDirectoy() throws Exception { String dir = "dir" + File.separator + "docs"; File dirFile = new File(ClassLoader.getSystemClassLoader().getResource(dir).getPath()); assertTrue(dir + " does not exist: " + dirFile.getAbsolutePath(), dirFile.exists()); Path input = new Path(tempDir, dir); for (File file : dirFile.listFiles()) { if (!file.isDirectory()) { Path filePath = new Path(input, file.getName()); addContentToFS(filePath, Files.toByteArray(file)); } } String jobName = "testDirectoy"; String[] args = new JobArgs().withJobName(jobName).withClassname(DirectoryIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString() + "/*").getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); MockRecordWriter writer = IngestJobMockMapRedOutFormat.writers.get(jobName); assertNotNull(writer); assertEquals(7, writer.map.size()); }
@Test public void testZip() throws Exception { String zip = "zip/zipData.zip"; File zipFile = new File(ClassLoader.getSystemClassLoader().getResource(zip).getPath()); assertTrue(zip + " does not exist: " + zipFile.getAbsolutePath(), zipFile.exists()); Path input = new Path(tempDir, zip); addContentToFS(input, Files.toByteArray(zipFile)); String jobName = "testZip"; String[] args = new JobArgs().withJobName(jobName).withClassname(ZipIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 6, new String[]{"test0.pdf", "test1.doc", "test0.doc", "test3.pdf", "test2.pdf", "test1.pdf"}); }
@Test public void testCSVLWS592() throws Exception { String csv = "csv/LWS592.csv"; File csvFile = new File(ClassLoader.getSystemClassLoader().getResource(csv).getPath()); assertTrue(csv + " does not exist: " + csvFile.getAbsolutePath(), csvFile.exists()); Path input = new Path(tempDir, csv); addContentToFS(input, Files.toString(csvFile, Charsets.UTF_8)); String jobName = "testCSVLWS592"; String[] args = new JobArgs().withJobName(jobName) .withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()) .withConf("csvFieldMapping[0=id,1=name_s,2=place_s];" + "csvFirstLineComment[false]") .withDArgs("-DcsvDelimiter=\u0001") .getJobArgs(); System.err.println(Arrays.toString(args)); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 2, new String[]{"2"}); }
@Test public void testCSVquoteswithCircumflex() throws Exception { String csv = "csv/quotes_with_circumflex.csv"; File csvFile = new File(ClassLoader.getSystemClassLoader().getResource(csv).getPath()); assertTrue(csv + " does not exist: " + csvFile.getAbsolutePath(), csvFile.exists()); Path input = new Path(tempDir, csv); addContentToFS(input, Files.toString(csvFile, Charsets.UTF_8)); String jobName = "testCSVquoteswithCircumflex"; String[] args = new JobArgs().withJobName(jobName).withClassname(CSVIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()) .withConf("csvDelimiter[^];csvFieldMapping[0=id,1=name_s];csvFirstLineComment[false]") .getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 2, new String[]{"2"}); }
@Test public void testWarc() throws Exception { String warc = "warc/at.warc"; File warcFile = new File(ClassLoader.getSystemClassLoader().getResource(warc).getPath()); assertTrue(warc + " does not exist: " + warcFile.getAbsolutePath(), warcFile.exists()); Path input = new Path(tempDir, warc); addContentToFS(input, Files.toString(warcFile, Charsets.UTF_8)); String jobName = "testWarc"; String[] args = new JobArgs().withJobName(jobName).withClassname(WarcIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 3, new String[]{"<urn:uuid:b328f1fe-b2ee-45c0-9139-908850810b52>", "<urn:uuid:6ee9accb-a284-47ef-8785-ed28aee2f79e>"}, "warc.WARC-Target-URI", "warc.WARC-Warcinfo-ID"); }
@Test public void testSolrXML() throws Exception { String solr = "sequence" + File.separator + "frankenstein_text_solr.seq"; File solrFile = new File(ClassLoader.getSystemClassLoader().getResource(solr).getPath()); assertTrue(solr + " does not exist: " + solrFile.getAbsolutePath(), solrFile.exists()); Path input = new Path(tempDir, solr); addContentToFS(input, Files.toByteArray(solrFile)); String jobName = "testSolrXML"; String[] args = new JobArgs().withJobName(jobName).withClassname(SolrXMLIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 776, new String[]{"solr_521", "solr_137", "solr_519"}, "body"); }
@Test public void testSequenceFile() throws Exception { String seq = "sequence" + File.separator + "frankenstein_text_text.seq"; File seqFile = new File(ClassLoader.getSystemClassLoader().getResource(seq).getPath()); assertTrue(seq + " does not exist: " + seqFile.getAbsolutePath(), seqFile.exists()); Path input = new Path(tempDir, seq); addContentToFS(input, Files.toByteArray(seqFile)); String jobName = "testSequenceFile"; String[] args = new JobArgs().withJobName(jobName).withClassname(SequenceFileIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(input.toUri().toString()).getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); verifyJob(jobName, 776, new String[]{"frank_seq_558", "frank_seq_171", "frank_seq_554", "frank_seq_551"}); }
@Test public void testRegex() throws Exception { String regex1 = "regex" + File.separator + "regex-small.txt"; File regexFile1 = new File(ClassLoader.getSystemClassLoader().getResource(regex1).getPath()); assertTrue(regex1 + " does not exist: " + regexFile1.getAbsolutePath(), regexFile1.exists()); Path input1 = new Path(tempDir, regex1); addContentToFS(input1, Files.toByteArray(regexFile1)); String regex2 = "regex" + File.separator + "regex-small-2.txt"; File regexFile2 = new File(ClassLoader.getSystemClassLoader().getResource(regex2).getPath()); assertTrue(regex2 + " does not exist: " + regexFile2.getAbsolutePath(), regexFile2.exists()); Path input2 = new Path(tempDir, regex2); addContentToFS(input2, Files.toByteArray(regexFile2)); String jobName = "testRegex"; String[] args = new JobArgs().withJobName(jobName).withClassname(RegexIngestMapper.class.getName()) .withCollection(DEFAULT_COLLECTION).withZkString(getBaseUrl()) .withInput(tempDir.toUri().toString() + File.separator + "regex" + File.separator + "regex-small*") .withDArgs("-D" + RegexIngestMapper.REGEX + "=\\w+", "-D" + RegexIngestMapper .GROUPS_TO_FIELDS + "=0=match") .getJobArgs(); int val = ToolRunner.run(conf, new IngestJob(), args); assertEquals(0, val); MockRecordWriter mockRecordWriter = IngestJobMockMapRedOutFormat.writers.get(jobName); Assert.assertNotNull(mockRecordWriter); assertEquals(2, mockRecordWriter.map.size()); }