@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 "); }
@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; } }
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; } } }
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; } }  }
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); }
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; }