温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

Nutch如何解析Html文档

发布时间:2021-10-12 16:03:18 来源:亿速云 阅读:110 作者:柒染 栏目:云计算

这篇文章给大家介绍Nutch如何解析Html文档,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。

解析Html文档 MapReduce任务描述

一、主程序调用

ParseSegment parseSegment = new ParseSegment(getConf());

if (!Fetcher.isParsing(job)) {

parseSegment.parse(segs[0]); // parse it, if needed

}

(1)、isParseing方法的实现

public static boolean isParsing(Configuration conf) {

return conf.getBoolean("fetcher.parse", true);

}

(2)、参数segs[0]

Path[] segs = generator.generate(

crawlDb, 

segments, 

-1, 

topN,

System.currentTimeMillis());

generate方法中 generatedSegments的生成过程。

这里疑点比较多,先放一放

// read the subdirectories generated in the temp 读取temp中生成的子文件夹

// output and turn them into segments 输出并且把他们转到segments中

//1、创建Path的集合对象

List<Path> generatedSegments = new ArrayList<Path>();

//2、

FileStatus[] status = fs.listStatus(tempDir);// 这里读取上面生成的多个fetchlist的segment

try {

for (FileStatus stat : status) {

Path subfetchlist = stat.getPath();

if (!subfetchlist.getName().startsWith("fetchlist-"))

continue;// 过滤不是以fetchlist-开头的文件

// start a new partition job for this segment

Path newSeg = partitionSegment(fs, segments, subfetchlist,

numLists);

// 对segment进行Partition操作,产生一个新的目录

generatedSegments.add(newSeg);

}

} catch (Exception e) {

LOG.warn("Generator: exception while partitioning segments, exiting ...");

fs.delete(tempDir, true);

return null;

}

二、job任务配置

job.setInputFormat(SequenceFileInputFormat.class);

job.setMapperClass(ParseSegment.class);

job.setReducerClass(ParseSegment.class);    

FileOutputFormat.setOutputPath(job, segment);

job.setOutputFormat(ParseOutputFormat.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(ParseImpl.class);

JobClient.runJob(job);

三、Map、reduce任务的输入和输出

map任务输入和输出

输入: WritableComparable/ Content

输出: Text/ ParseImpl

  public void map(WritableComparable<?> key, Content content,

                  OutputCollector<Text, ParseImpl> output, Reporter reporter)

reduce任务输入和输出

输入: Text/Iterator<Writable>

输出: Text/Writable

  public void reduce(Text key, Iterator<Writable> values,

                     OutputCollector<Text, Writable> output, Reporter reporter)

四、job任务输入类SequenceFileInputFormat

  protected FileStatus[] listStatus(JobConf job) throws IOException {

    FileStatus[] files = super.listStatus(job);

    for (int i = 0; i < files.length; i++) {

      FileStatus file = files[i];

      if (file.isDir()) {     // it's a MapFile

        Path dataFile = new Path(file.getPath(), MapFile.DATA_FILE_NAME);

        FileSystem fs = file.getPath().getFileSystem(job);

        // use the data file

        files[i] = fs.getFileStatus(dataFile);

      }

    }

    return files;

  }

  public RecordReader<K, V> getRecordReader(InputSplit split,

                                      JobConf job, Reporter reporter)

    throws IOException {

    reporter.setStatus(split.toString());

    return new SequenceFileRecordReader<K, V>(job, (FileSplit) split);

  }

五、map()方法和reduce()方法中的实现

(1)、map任务 

org.apache.nutch.parse.ParseSegment

  public void map(WritableComparable<?> key, Content content,

                  OutputCollector<Text, ParseImpl> output, Reporter reporter)

    throws IOException {

    // convert on the fly from old UTF8 keys

    if (key instanceof Text) {

      newKey.set(key.toString());

      key = newKey;

    }

//2、 获取抓取状态,

//Nutch.FETCH_STATUS_KEY)——> _fst_

    int status =

      Integer.parseInt(content.getMetadata().get(Nutch.FETCH_STATUS_KEY));

//3、如果成功抓取,如果没有抓取成功,就跳过这条记录

    if (status != CrawlDatum.STATUS_FETCH_SUCCESS) {

      // content not fetched successfully, skip document

      LOG.debug("Skipping " + key + " as content is not fetched successfully");

      return;

    }

     //4、判断是否试过截断,文档中是否被截断,如果要跳过截断,且文档是被截断了,也跳过这条记录

    if (skipTruncated && isTruncated(content)) {

      return;

    }

    ParseResult parseResult = null;

    try {

      //5、创建一个ParseUtil对象,调用解析方法parse,并返回一个解析结果ParseResult

parseResult = new ParseUtil(getConf()).parse(content);

    } catch (Exception e) {

      LOG.warn("Error parsing: " + key + ": " +StringUtils.stringifyException(e));

      return;

    }

//以上主要是解析,一下是对解析的处理

//————————————————————————————————————————————————————————————————————————————————

//6、遍历上一步解析得到的解析结果,

    for (Entry<Text, Parse> entry : parseResult) {

      //7、获取键和值

Text url = entry.getKey();

      Parse parse = entry.getValue();

//8、获取状态      

ParseStatus parseStatus = parse.getData().getStatus();

      long start = System.currentTimeMillis();

//9、计数器+1

      reporter.incrCounter("ParserStatus",     ParseStatus.majorCodes[parseStatus.getMajorCode()], 1);

//10、如果解析不成功,将parse持有的所有对象都置为null。

      if (!parseStatus.isSuccess()) {

        LOG.warn("Error parsing: " + key + ": " + parseStatus);

        parse = parseStatus.getEmptyParse(getConf());

      }

      // pass segment name to parse data

//11、将segment名称赋值给parse data

      parse.getData().getContentMeta().set(Nutch.SEGMENT_NAME_KEY, 

                                           getConf().get(Nutch.SEGMENT_NAME_KEY));

      // compute the new signature

//12、计算新的分值

      byte[] signature = 

        SignatureFactory.getSignature(getConf()).calculate(content, parse); 

//13、设置digest的值

parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY, 

          StringUtil.toHexString(signature));

      try {

        scfilters.passScoreAfterParsing(url, content, parse);

      } catch (ScoringFilterException e) {

        if (LOG.isWarnEnabled()) {

          LOG.warn("Error passing score: "+ url +": "+e.getMessage());

        }

      }

      long end = System.currentTimeMillis();

      LOG.info("Parsed (" + Long.toString(end - start) + "ms):" + url);

      output.collect(url, new ParseImpl(new ParseText(parse.getText()), 

                                        parse.getData(), parse.isCanonical()));

    }

  }

ParseResult对象的特点

a、实现了Iterable接口 ,可迭代;迭代对象是Entry,entry的key是Text,值是Parse,如下

public class ParseResult implements Iterable<Map.Entry<Text, Parse>> 

迭代方法如下,

  public Iterator<Entry<Text, Parse>> iterator() {

    return parseMap.entrySet().iterator();

  }

可以看到是使用了Map结合的迭代器

b、持有一个HashMap用于存放解析结果,另持有当前的url,代码如下

private Map<Text, Parse> parseMap;

private String originalUrl;

Parse和ParseImpl的说明

Parse是一个接口,其中的3个方法如下

/** The textual(正文) content of the page. This is indexed, searched, and used when  generating snippets.*/ 

//网页正文内容,将被索引,搜索,生成快照

String getText();

/** Other data extracted from the page. */

//冲网页中提取的其他数据

ParseData getData();

/** Indicates if the parse is coming from a url or a sub-url */

//标识这个Parse是否来自于url或者一个子url

boolean isCanonical();//Canonical:正规的

ParseImpl实现了Parse和Writable接口

ParseImpl中有3个字段,如下所示,其中isCanonical是在构造的时候传入的,默认是true

  private ParseText text;

  private ParseData data;

  private boolean isCanonical;//规则

用于去重的digest的计算

//12、计算新的分值

      byte[] signature = 

        SignatureFactory.getSignature(getConf()).calculate(content, parse); 

//13、设置digest的值

parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY, 

          StringUtil.toHexString(signature));

Signature:n.签名; 署名; 识别标志,鲜明特征; [医]药的用法说明;

SignatureFactory类中的getSignature方法

该方法看ObjectCache中有没有Signature的实现,如果没有就利用反射创建一个并返回。

  /** Return the default Signature implementation. */

  public static Signature getSignature(Configuration conf) {

    String clazz = conf.get("db.signature.class", MD5Signature.class.getName());

    ObjectCache objectCache = ObjectCache.get(conf);

    Signature impl = (Signature)objectCache.getObject(clazz);

    if (impl == null) {

      try {

        if (LOG.isInfoEnabled()) {

          LOG.info("Using Signature impl: " + clazz);

        }

        Class<?> implClass = Class.forName(clazz);

        impl = (Signature)implClass.newInstance();

        impl.setConf(conf);

        objectCache.setObject(clazz, impl);

      } catch (Exception e) {

        throw new RuntimeException("Couldn't create " + clazz, e);

      }

    }

    return impl;

  }

***** 重要

计算网页特征,最后是调用了Signature的calculate方法,以下是Signature实现类MD5Signaure的类代码

/**

 * Default implementation of a page signature. 

默认的Signature的实现类

 * It calculates an MD5 hash of the raw binary content of a page. 

它计算page内容的原始的二进制代码的md5 hash值

 * In case there is no content, 

 * it calculates a hash from the page's URL.

 * 

 * @author Andrzej Bialecki &lt;ab@getopt.org&gt;

 */

public class MD5Signature extends Signature {

  public byte[] calculate(Content content, Parse parse) {

    byte[] data = content.getContent();

    if (data == null) data = content.getUrl().getBytes();

    return MD5Hash.digest(data).getDigest();

  }

}

(2)、reduce任务

未完待续

多线程解析

//解析是多线程地 

  private ParseResult runParser(Parser p, Content content) {

    ParseCallable pc = new ParseCallable(p, content);

    Future<ParseResult> task = executorService.submit(pc);

    ParseResult res = null;

    try {

      res = task.get(maxParseTime, TimeUnit.SECONDS);

    } catch (Exception e) {

      LOG.warn("Error parsing " + content.getUrl() + " with " + p, e);

      task.cancel(true);

    } finally {

      pc = null;

    }

    return res;

  }

关于Nutch如何解析Html文档就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI