Nutch1.3集成Solr网页快照功能实现（一）

  public void indexSolr(String solrUrl, Path crawlDb, Path linkDb,

      List<Path> segments) throws IOException {

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    long start = System.currentTimeMillis();

    LOG.info("SolrIndexer: starting at " + sdf.format(start));

    final JobConf job = new NutchJob(getConf());

    job.setJobName("index-solr " + solrUrl);

    IndexerMapReduce.initMRJob(crawlDb, linkDb, segments, job);

    job.set(SolrConstants.SERVER_URL, solrUrl);

    NutchIndexWriterFactory.addClassToConf(job, SolrWriter.class);

    job.setReduceSpeculativeExecution(false);

    final Path tmp = new Path("tmp_" + System.currentTimeMillis() + "-" +

                         new Random().nextInt());

    FileOutputFormat.setOutputPath(job, tmp);

    try {

      JobClient.runJob(job);

      // do the commits once and for all the reducers in one go

      SolrServer solr =  new CommonsHttpSolrServer(solrUrl);

      solr.commit();

      long end = System.currentTimeMillis();

      LOG.info("SolrIndexer: finished at " + sdf.format(end) + ", elapsed: " + TimingUtil.elapsedTime(start, end));

    }

    catch (Exception e){

      LOG.error(e);

    } finally {

      FileSystem.get(job).delete(tmp, true);

    }

  }

  public static void initMRJob(Path crawlDb, Path linkDb,

                           Collection<Path> segments,

                           JobConf job) {

    LOG.info("IndexerMapReduce: crawldb: " + crawlDb);

    LOG.info("IndexerMapReduce: linkdb: " + linkDb);

    for (final Path segment : segments) {

      LOG.info("IndexerMapReduces: adding segment: " + segment);

      FileInputFormat.addInputPath(job, new Path(segment, CrawlDatum.FETCH_DIR_NAME));

      FileInputFormat.addInputPath(job, new Path(segment, CrawlDatum.PARSE_DIR_NAME));

      FileInputFormat.addInputPath(job, new Path(segment, ParseData.DIR_NAME));

      FileInputFormat.addInputPath(job, new Path(segment, ParseText.DIR_NAME));

    }

    FileInputFormat.addInputPath(job, new Path(crawlDb, CrawlDb.CURRENT_NAME));

    FileInputFormat.addInputPath(job, new Path(linkDb, LinkDb.CURRENT_NAME));

    job.setInputFormat(SequenceFileInputFormat.class);

    job.setMapperClass(IndexerMapReduce.class);

    job.setReducerClass(IndexerMapReduce.class);

    job.setOutputFormat(IndexerOutputFormat.class);

    job.setOutputKeyClass(Text.class);

    job.setMapOutputValueClass(NutchWritable.class);

    job.setOutputValueClass(NutchWritable.class);

  }