parsesegment.java

来自「一些简要的公爵类一些简要的公爵类一些简要的公爵类」· Java 代码 · 共 617 行 · 第 1/2 页

    }

    public final void write(DataOutput out) throws IOException {
      super.write(out);                             // write version
      out.writeByte(status);
      parseData.write(out);
      parseText.write(out);
      return;
    }
  }
			
  /**
   * ParseSegment constructor
   */
  public ParseSegment(NutchFileSystem nfs, String directory, boolean dryRun)
    throws IOException {

    File file;

    this.nfs = nfs;
    this.directory = directory;
    this.dryRun = dryRun;

    // FetcherOutput.DIR_NAME_NP must exist
    file = new File(directory, FetcherOutput.DIR_NAME_NP);
    if (!nfs.exists(file))
      throw new IOException("Directory missing: "+FetcherOutput.DIR_NAME_NP);

    if (dryRun)
      return;

    // clean old FetcherOutput.DIR_NAME
    file = new File(directory, FetcherOutput.DIR_NAME);
    if (nfs.exists(file)) {
      LOG.info("Deleting old "+file.getName());
      nfs.delete(file);
    }

    // clean old unsortedFile
    this.unsortedFile = new File(directory, ParserOutput.DIR_NAME+".unsorted");
    if (nfs.exists(this.unsortedFile)) {
      LOG.info("Deleting old "+this.unsortedFile.getName());
      nfs.delete(this.unsortedFile);
    }

    // clean old sortedFile
    this.sortedFile = new File(directory, ParserOutput.DIR_NAME+".sorted");
    if (nfs.exists(this.sortedFile)) {
      LOG.info("Deleting old "+this.sortedFile.getName());
      nfs.delete(this.sortedFile);
    }

    // clean old ParseData.DIR_NAME
    file = new File(directory, ParseData.DIR_NAME);
    if (nfs.exists(file)) {
      LOG.info("Deleting old "+file.getName());
      nfs.delete(file);
    }

    // clean old ParseText.DIR_NAME
    file = new File(directory, ParseText.DIR_NAME);
    if (nfs.exists(file)) {
      LOG.info("Deleting old "+file.getName());
      nfs.delete(file);
    }

  }

  /** Set thread count */
  public void setThreadCount(int threadCount) {
    this.threadCount=threadCount;
  }

  /** Set the logging level. */
  public static void setLogLevel(Level level) {
    LOG.setLevel(level);
    PluginRepository.LOG.setLevel(level);
    ParserFactory.LOG.setLevel(level);
    LOG.info("logging at " + level);
  }

  /** Set if clean intermediates. */
  public void setClean(boolean clean) {
    this.clean = clean;
  }

  /** Display the status of the parser run. */
  public void status() {
    long ms = System.currentTimeMillis() - start;
    LOG.info("status: "
             + pages + " pages, "
             + errors + " errors, "
             + bytes + " bytes, "
             + ms + " ms");
    LOG.info("status: "
             + (((float)pages)/(ms/1000.0f))+" pages/s, "
             + (((float)bytes*8/1024)/(ms/1000.0f))+" kb/s, "
             + (((float)bytes)/pages) + " bytes/page");
  }

  /** Parse contents by multiple threads and save as unsorted ParserOutput */
  public void parse() throws IOException, InterruptedException {

    fetcherNPReader = new ArrayFile.Reader
      (nfs, (new File(directory, FetcherOutput.DIR_NAME_NP)).getPath());
    contentReader = new ArrayFile.Reader
      (nfs, (new File(directory, Content.DIR_NAME)).getPath());

    if (!this.dryRun) {
      parserOutputWriter = new SequenceFile.Writer
        (nfs, unsortedFile.getPath(), LongWritable.class, ParserOutput.class);
    }

    start = System.currentTimeMillis();

    for (int i = 0; i < threadCount; i++) {       // spawn threads
      ParserThread thread = new ParserThread(); 
      thread.start();
    }

    do {
      Thread.sleep(1000);

      if (LogFormatter.hasLoggedSevere()) 
        throw new RuntimeException("SEVERE error logged.  Exiting parser.");

    } while (group.activeCount() > 0);            // wait for threads to finish

    fetcherNPReader.close();
    contentReader.close();
    if (!this.dryRun)
      parserOutputWriter.close();

    status();                                     // print final status
  }

  /** Sort ParserOutput */
  public void sort() throws IOException {

    if (this.dryRun)
      return;

    LOG.info("Sorting ParserOutput");

    start = System.currentTimeMillis();

    SequenceFile.Sorter sorter = new SequenceFile.Sorter
      (nfs, new LongWritable.Comparator(), ParserOutput.class);

    sorter.sort(unsortedFile.getPath(), sortedFile.getPath());

    double localSecs = (System.currentTimeMillis() - start) / 1000.0;
    LOG.info("Sorted: " + (pages+errors) + " entries in " + localSecs + "s, "
      + ((pages+errors)/localSecs) + " entries/s");

    if (this.clean) {
      LOG.info("Deleting intermediate "+unsortedFile.getName());
      nfs.delete(unsortedFile);
    }

    return;
  }

  /**
   * Split sorted ParserOutput into ParseData and ParseText,
   * and generate new FetcherOutput with updated status
   */
  public void save() throws IOException {

    if (this.dryRun)
      return;

    LOG.info("Saving ParseData and ParseText separately");

    start = System.currentTimeMillis();

    SequenceFile.Reader parserOutputReader
      = new SequenceFile.Reader(nfs, sortedFile.getPath());

    ArrayFile.Reader fetcherNPReader = new ArrayFile.Reader(nfs,
      (new File(directory, FetcherOutput.DIR_NAME_NP)).getPath());

    ArrayFile.Writer fetcherWriter = new ArrayFile.Writer(nfs,
      (new File(directory, FetcherOutput.DIR_NAME)).getPath(),
      FetcherOutput.class);

    ArrayFile.Writer parseDataWriter = new ArrayFile.Writer(nfs,
      (new File(directory, ParseData.DIR_NAME)).getPath(), ParseData.class);
    ArrayFile.Writer parseTextWriter = new ArrayFile.Writer(nfs,
      (new File(directory, ParseText.DIR_NAME)).getPath(), ParseText.class);

    try {
      LongWritable key = new LongWritable();
      ParserOutput val = new ParserOutput();
      FetcherOutput fo = new FetcherOutput();
      int count = 0;
      int status;
      while (parserOutputReader.next(key,val)) {
        fetcherNPReader.next(fo);
        // safe guarding
        if (fetcherNPReader.key() != key.get())
          throw new IOException("Mismatch between entries under "
            + FetcherOutput.DIR_NAME_NP + " and in " + sortedFile.getName());
        // reset status in fo (FetcherOutput), using status in ParserOutput
        switch (val.getStatus()) {
        case ParserOutput.SUCCESS:
          fo.setStatus(FetcherOutput.SUCCESS);
          break;
        case ParserOutput.UNKNOWN:
        case ParserOutput.FAILURE:
          fo.setStatus(FetcherOutput.CANT_PARSE);
          break;
        case ParserOutput.NOFETCH:
        default:
          // do not reset
        }
        fetcherWriter.append(fo);
        parseDataWriter.append(val.getParseData());
        parseTextWriter.append(val.getParseText());
        count++;
      }
      // safe guard! make sure there are identical entries
      // in (fetcher, content) and in (parseData, parseText)
      if (count != (pages+errors))
        throw new IOException("Missing entries: expect "+(pages+errors)
          +", but have "+count+" entries instead.");
    } finally {
      fetcherNPReader.close();
      fetcherWriter.close();
      parseDataWriter.close();
      parseTextWriter.close();
      parserOutputReader.close();
    }

    double localSecs = (System.currentTimeMillis() - start) / 1000.0;
    LOG.info("Saved: " + (pages+errors) + " entries in " + localSecs + "s, "
      + ((pages+errors)/localSecs) + " entries/s");

    if (this.clean) {
      LOG.info("Deleting intermediate "+sortedFile.getName());
      nfs.delete(sortedFile);
    }

    return;
  }

  /** main method */
  public static void main(String[] args) throws Exception {
    int threadCount = -1;
    boolean showThreadID = false;
    boolean dryRun = false;
    String logLevel = "info";
    boolean clean = true;
    String directory = null;

    String usage = "Usage: ParseSegment (-local | -ndfs <namenode:port>) [-threads n] [-showThreadID] [-dryRun] [-logLevel level] [-noClean] dir";

    if (args.length == 0) {
      System.err.println(usage);
      System.exit(-1);
    }
      
    // parse command line
    NutchFileSystem nfs = NutchFileSystem.parseArgs(args, 0);

    for (int i = 0; i < args.length; i++) {
      if (args[i] == null) {
          continue;
      } else if (args[i].equals("-threads")) {
        threadCount =  Integer.parseInt(args[++i]);
      } else if (args[i].equals("-showThreadID")) {
        showThreadID = true;
      } else if (args[i].equals("-dryRun")) {
        dryRun = true;
      } else if (args[i].equals("-logLevel")) {
        logLevel = args[++i];
      } else if (args[i].equals("-noClean")) {
        clean = false;
      } else {
        directory = args[i];
      }
    }

    try {

      ParseSegment parseSegment = new ParseSegment(nfs, directory, dryRun);

      parseSegment.setLogLevel
        (Level.parse((new String(logLevel)).toUpperCase()));

      if (threadCount != -1)
        parseSegment.setThreadCount(threadCount);
      if (showThreadID)
        LogFormatter.setShowThreadIDs(showThreadID);

      parseSegment.setClean(clean);

      parseSegment.parse();
      parseSegment.sort();
      parseSegment.save();

    } finally {
      nfs.close();
    }

  }
}