(Map<String, Object> args)
| 162 | } |
| 163 | |
| 164 | @Override |
| 165 | public Map<String, Object> run(Map<String, Object> args) throws Exception { |
| 166 | checkConfiguration(); |
| 167 | String batchId = (String) args.get(Nutch.ARG_BATCH); |
| 168 | Integer threads = (Integer) args.get(Nutch.ARG_THREADS); |
| 169 | Boolean shouldResume = (Boolean) args.get(Nutch.ARG_RESUME); |
| 170 | Integer numTasks = (Integer) args.get(Nutch.ARG_NUMTASKS); |
| 171 | Boolean stmDetect = (Boolean) args.get(Nutch.ARG_SITEMAP_DETECT); |
| 172 | Boolean sitemap = (Boolean) args.get(Nutch.ARG_SITEMAP); |
| 173 | |
| 174 | if (threads != null && threads > 0) { |
| 175 | getConf().setInt(THREADS_KEY, threads); |
| 176 | } |
| 177 | if (batchId == null) { |
| 178 | batchId = Nutch.ALL_BATCH_ID_STR; |
| 179 | } |
| 180 | getConf().set(GeneratorJob.BATCH_ID, batchId); |
| 181 | if (shouldResume != null) { |
| 182 | getConf().setBoolean(RESUME_KEY, shouldResume); |
| 183 | } |
| 184 | if (stmDetect != null) { |
| 185 | getConf().setBoolean(SITEMAP_DETECT, stmDetect); |
| 186 | } |
| 187 | if (sitemap != null) { |
| 188 | getConf().setBoolean(SITEMAP, sitemap); |
| 189 | } |
| 190 | |
| 191 | LOG.info("FetcherJob: threads: {}", getConf().getInt(THREADS_KEY, 10)); |
| 192 | LOG.info("FetcherJob: parsing: {}", getConf().getBoolean(PARSE_KEY, false)); |
| 193 | LOG.info("FetcherJob: resuming: {}", getConf().getBoolean(RESUME_KEY, false)); |
| 194 | |
| 195 | // set the actual time for the timelimit relative |
| 196 | // to the beginning of the whole job and not of a specific task |
| 197 | // otherwise it keeps trying again if a task fails |
| 198 | long timelimit = getConf().getLong("fetcher.timelimit.mins", -1); |
| 199 | if (timelimit != -1) { |
| 200 | timelimit = System.currentTimeMillis() + (timelimit * 60 * 1000); |
| 201 | getConf().setLong("fetcher.timelimit", timelimit); |
| 202 | } |
| 203 | LOG.info("FetcherJob : timelimit set for : {}", getConf().getLong("fetcher.timelimit", -1)); |
| 204 | numJobs = 1; |
| 205 | currentJob = NutchJob.getInstance(getConf(), "fetch"); |
| 206 | |
| 207 | // for politeness, don't permit parallel execution of a single task |
| 208 | currentJob.setReduceSpeculativeExecution(false); |
| 209 | |
| 210 | Collection<WebPage.Field> fields = getFields(currentJob); |
| 211 | MapFieldValueFilter<String, WebPage> batchIdFilter = getBatchIdFilter(batchId); |
| 212 | StorageUtils.initMapperJob(currentJob, fields, IntWritable.class, |
| 213 | FetchEntry.class, FetcherMapper.class, FetchEntryPartitioner.class, |
| 214 | batchIdFilter, false); |
| 215 | StorageUtils.initReducerJob(currentJob, FetcherReducer.class); |
| 216 | if (numTasks == null || numTasks < 1) { |
| 217 | currentJob.setNumReduceTasks(currentJob.getConfiguration().getInt( |
| 218 | "mapreduce.job.reduces", currentJob.getNumReduceTasks())); |
| 219 | } else { |
| 220 | currentJob.setNumReduceTasks(numTasks); |
| 221 | } |
no test coverage detected