diff --git a/src/java/org/apache/nutch/crawl/CrawlDb.java b/src/java/org/apache/nutch/crawl/CrawlDb.java index 01598a5f18..32081e1d61 100644 --- a/src/java/org/apache/nutch/crawl/CrawlDb.java +++ b/src/java/org/apache/nutch/crawl/CrawlDb.java @@ -43,6 +43,7 @@ import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apache.nutch.metadata.Nutch; +import org.apache.nutch.metrics.NutchMetrics; import org.apache.nutch.util.FSUtils; import org.apache.nutch.util.HadoopFSUtil; import org.apache.nutch.util.LockUtil; @@ -145,7 +146,7 @@ public void update(Path crawlDb, Path[] segments, boolean normalize, if (filter) { long urlsFiltered = job.getCounters() - .findCounter("CrawlDB filter", "URLs filtered").getValue(); + .findCounter(NutchMetrics.GROUP_CRAWLDB_FILTER, NutchMetrics.CRAWLDB_URLS_FILTERED_TOTAL).getValue(); LOG.info( "CrawlDb update: Total number of existing URLs in CrawlDb rejected by URL filters: {}", urlsFiltered); diff --git a/src/java/org/apache/nutch/crawl/DeduplicationJob.java b/src/java/org/apache/nutch/crawl/DeduplicationJob.java index cdb291fe85..d5f983a273 100644 --- a/src/java/org/apache/nutch/crawl/DeduplicationJob.java +++ b/src/java/org/apache/nutch/crawl/DeduplicationJob.java @@ -335,12 +335,10 @@ public int run(String[] args) throws IOException { fs.delete(tempDir, true); throw new RuntimeException(message); } - CounterGroup g = job.getCounters().getGroup("DeduplicationJobStatus"); - if (g != null) { - Counter counter = g.findCounter("Documents marked as duplicate"); - long dups = counter.getValue(); - LOG.info("Deduplication: {} documents marked as duplicates", dups); - } + long dups = job.getCounters() + .findCounter(NutchMetrics.GROUP_DEDUP, NutchMetrics.DEDUP_DOCUMENTS_MARKED_DUPLICATE_TOTAL) + .getValue(); + LOG.info("Deduplication: {} documents marked as duplicates", dups); } catch (IOException | InterruptedException | ClassNotFoundException e) { LOG.error("DeduplicationJob:", e); fs.delete(tempDir, true); diff --git a/src/java/org/apache/nutch/fetcher/QueueFeeder.java b/src/java/org/apache/nutch/fetcher/QueueFeeder.java index 6ee973dd3b..5dfa24fd06 100644 --- a/src/java/org/apache/nutch/fetcher/QueueFeeder.java +++ b/src/java/org/apache/nutch/fetcher/QueueFeeder.java @@ -22,6 +22,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Counter; import org.apache.nutch.crawl.CrawlDatum; import org.apache.nutch.fetcher.FetchItemQueues.QueuingStatus; import org.apache.nutch.fetcher.Fetcher.FetcherRun; @@ -48,6 +49,12 @@ public class QueueFeeder extends Thread { private URLNormalizers urlNormalizers = null; private String urlNormalizerScope = URLNormalizers.SCOPE_DEFAULT; + // Cached counter references to avoid repeated lookups in hot paths + private Counter hitByTimeoutCounter; + private Counter hitByTimelimitCounter; + private Counter filteredCounter; + private Counter aboveExceptionThresholdCounter; + public QueueFeeder(FetcherRun.Context context, FetchItemQueues queues, int size) { this.context = context; @@ -62,6 +69,21 @@ public QueueFeeder(FetcherRun.Context context, if (conf.getBoolean("fetcher.normalize.urls", false)) { urlNormalizers = new URLNormalizers(conf, urlNormalizerScope); } + initCounters(); + } + + /** + * Initialize cached counter references to avoid repeated lookups in hot paths. + */ + private void initCounters() { + hitByTimeoutCounter = context.getCounter( + NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_HIT_BY_TIMEOUT_TOTAL); + hitByTimelimitCounter = context.getCounter( + NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_HIT_BY_TIMELIMIT_TOTAL); + filteredCounter = context.getCounter( + NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_FILTERED_TOTAL); + aboveExceptionThresholdCounter = context.getCounter( + NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_ABOVE_EXCEPTION_THRESHOLD_TOTAL); } /** Filter and normalize the url */ @@ -95,16 +117,14 @@ public void run() { LOG.info("QueueFeeder stopping, timeout reached."); } queuingStatus[qstatus]++; - context.getCounter(NutchMetrics.GROUP_FETCHER, - NutchMetrics.FETCHER_HIT_BY_TIMEOUT_TOTAL).increment(1); + hitByTimeoutCounter.increment(1); } else { int qstatus = QueuingStatus.HIT_BY_TIMELIMIT.ordinal(); if (queuingStatus[qstatus] == 0) { LOG.info("QueueFeeder stopping, timelimit exceeded."); } queuingStatus[qstatus]++; - context.getCounter(NutchMetrics.GROUP_FETCHER, - NutchMetrics.FETCHER_HIT_BY_TIMELIMIT_TOTAL).increment(1); + hitByTimelimitCounter.increment(1); } try { hasMore = context.nextKeyValue(); @@ -136,8 +156,7 @@ public void run() { String u = filterNormalize(url.toString()); if (u == null) { // filtered or failed to normalize - context.getCounter(NutchMetrics.GROUP_FETCHER, - NutchMetrics.FETCHER_FILTERED_TOTAL).increment(1); + filteredCounter.increment(1); continue; } url = new Text(u); @@ -154,8 +173,7 @@ public void run() { QueuingStatus status = queues.addFetchItem(url, datum); queuingStatus[status.ordinal()]++; if (status == QueuingStatus.ABOVE_EXCEPTION_THRESHOLD) { - context.getCounter(NutchMetrics.GROUP_FETCHER, - NutchMetrics.FETCHER_ABOVE_EXCEPTION_THRESHOLD_TOTAL).increment(1); + aboveExceptionThresholdCounter.increment(1); } cnt++; feed--; diff --git a/src/java/org/apache/nutch/hostdb/UpdateHostDbMapper.java b/src/java/org/apache/nutch/hostdb/UpdateHostDbMapper.java index 1495f74914..8de2dcdf2c 100644 --- a/src/java/org/apache/nutch/hostdb/UpdateHostDbMapper.java +++ b/src/java/org/apache/nutch/hostdb/UpdateHostDbMapper.java @@ -24,6 +24,7 @@ import org.apache.hadoop.io.FloatWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; +import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.conf.Configuration; @@ -61,6 +62,10 @@ public class UpdateHostDbMapper protected URLFilters filters = null; protected URLNormalizers normalizers = null; + // Cached counter references to avoid repeated lookups in hot paths + protected Counter malformedUrlCounter; + protected Counter filteredRecordsCounter; + @Override public void setup(Mapper.Context context) { Configuration conf = context.getConfiguration(); @@ -72,6 +77,12 @@ public void setup(Mapper.Context context) { filters = new URLFilters(conf); if (normalize) normalizers = new URLNormalizers(conf, URLNormalizers.SCOPE_DEFAULT); + + // Initialize cached counter references + malformedUrlCounter = context.getCounter( + NutchMetrics.GROUP_HOSTDB, NutchMetrics.HOSTDB_MALFORMED_URL_TOTAL); + filteredRecordsCounter = context.getCounter( + NutchMetrics.GROUP_HOSTDB, NutchMetrics.HOSTDB_FILTERED_RECORDS_TOTAL); } /** @@ -137,8 +148,7 @@ public void map(Text key, Writable value, try { url = new URL(keyStr); } catch (MalformedURLException e) { - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_MALFORMED_URL_TOTAL).increment(1); + malformedUrlCounter.increment(1); return; } String hostName = URLUtil.getHost(url); @@ -148,8 +158,7 @@ public void map(Text key, Writable value, // Filtered out? if (buffer == null) { - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_FILTERED_RECORDS_TOTAL).increment(1); + filteredRecordsCounter.increment(1); LOG.debug("UpdateHostDb: {} crawldatum has been filtered", hostName); return; } @@ -222,8 +231,7 @@ public void map(Text key, Writable value, // Filtered out? if (buffer == null) { - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_FILTERED_RECORDS_TOTAL).increment(1); + filteredRecordsCounter.increment(1); LOG.debug("UpdateHostDb: {} hostdatum has been filtered", keyStr); return; } @@ -247,8 +255,7 @@ public void map(Text key, Writable value, // Filtered out? if (buffer == null) { - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_FILTERED_RECORDS_TOTAL).increment(1); + filteredRecordsCounter.increment(1); LOG.debug("UpdateHostDb: {} score has been filtered", keyStr); return; } diff --git a/src/java/org/apache/nutch/hostdb/UpdateHostDbReducer.java b/src/java/org/apache/nutch/hostdb/UpdateHostDbReducer.java index 039fa5ba13..6c979f222e 100644 --- a/src/java/org/apache/nutch/hostdb/UpdateHostDbReducer.java +++ b/src/java/org/apache/nutch/hostdb/UpdateHostDbReducer.java @@ -31,6 +31,7 @@ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; +import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.util.StringUtils; @@ -73,6 +74,11 @@ public class UpdateHostDbReducer protected BlockingQueue queue = new SynchronousQueue<>(); protected ThreadPoolExecutor executor = null; + // Cached counter references to avoid repeated lookups in hot paths + protected Counter urlLimitNotReachedCounter; + protected Counter totalHostsCounter; + protected Counter skippedNotEligibleCounter; + /** * Configures the thread pool and prestarts all resolver threads. */ @@ -146,6 +152,14 @@ public void setup(Reducer.Context context) // Run all threads in the pool executor.prestartAllCoreThreads(); } + + // Initialize cached counter references + urlLimitNotReachedCounter = context.getCounter( + NutchMetrics.GROUP_HOSTDB, NutchMetrics.HOSTDB_URL_LIMIT_NOT_REACHED_TOTAL); + totalHostsCounter = context.getCounter( + NutchMetrics.GROUP_HOSTDB, NutchMetrics.HOSTDB_TOTAL_HOSTS_TOTAL); + skippedNotEligibleCounter = context.getCounter( + NutchMetrics.GROUP_HOSTDB, NutchMetrics.HOSTDB_SKIPPED_NOT_ELIGIBLE_TOTAL); } /** @@ -380,14 +394,12 @@ else if (value instanceof FloatWritable) { // Impose limits on minimum number of URLs? if (urlLimit > -1l) { if (hostDatum.numRecords() < urlLimit) { - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_URL_LIMIT_NOT_REACHED_TOTAL).increment(1); + urlLimitNotReachedCounter.increment(1); return; } } - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_TOTAL_HOSTS_TOTAL).increment(1); + totalHostsCounter.increment(1); // See if this record is to be checked if (shouldCheck(hostDatum)) { @@ -404,8 +416,7 @@ else if (value instanceof FloatWritable) { // Do not progress, the datum will be written in the resolver thread return; } else if (checkAny) { - context.getCounter(NutchMetrics.GROUP_HOSTDB, - NutchMetrics.HOSTDB_SKIPPED_NOT_ELIGIBLE_TOTAL).increment(1); + skippedNotEligibleCounter.increment(1); LOG.debug("UpdateHostDb: {}: skipped_not_eligible", key); } diff --git a/src/java/org/apache/nutch/indexer/IndexerMapReduce.java b/src/java/org/apache/nutch/indexer/IndexerMapReduce.java index 9086a19839..b61a7f99cd 100644 --- a/src/java/org/apache/nutch/indexer/IndexerMapReduce.java +++ b/src/java/org/apache/nutch/indexer/IndexerMapReduce.java @@ -30,6 +30,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; +import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; @@ -219,6 +220,18 @@ public static class IndexerReducer extends // Latency tracker for indexing timing metrics private LatencyTracker indexLatencyTracker; + // Cached counter references to avoid repeated lookups in hot paths + private Counter deletedRobotsNoIndexCounter; + private Counter deletedGoneCounter; + private Counter deletedRedirectsCounter; + private Counter deletedDuplicatesCounter; + private Counter skippedNotModifiedCounter; + private Counter errorsScoringFilterCounter; + private Counter errorsIndexingFilterCounter; + private Counter deletedByIndexingFilterCounter; + private Counter skippedByIndexingFilterCounter; + private Counter indexedCounter; + @Override public void setup(Reducer.Context context) { Configuration conf = context.getConfiguration(); @@ -247,6 +260,35 @@ public void setup(Reducer.Context c // Initialize latency tracker for indexing timing indexLatencyTracker = new LatencyTracker( NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_LATENCY); + + // Initialize cached counter references + initCounters(context); + } + + /** + * Initialize cached counter references to avoid repeated lookups in hot paths. + */ + private void initCounters(Reducer.Context context) { + deletedRobotsNoIndexCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_DELETED_ROBOTS_NOINDEX_TOTAL); + deletedGoneCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_DELETED_GONE_TOTAL); + deletedRedirectsCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_DELETED_REDIRECTS_TOTAL); + deletedDuplicatesCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_DELETED_DUPLICATES_TOTAL); + skippedNotModifiedCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_SKIPPED_NOT_MODIFIED_TOTAL); + errorsScoringFilterCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_ERRORS_SCORING_FILTER_TOTAL); + errorsIndexingFilterCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_ERRORS_INDEXING_FILTER_TOTAL); + deletedByIndexingFilterCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_DELETED_BY_INDEXING_FILTER_TOTAL); + skippedByIndexingFilterCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_SKIPPED_BY_INDEXING_FILTER_TOTAL); + indexedCounter = context.getCounter( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_INDEXED_TOTAL); } @Override @@ -299,8 +341,7 @@ public void reduce(Text key, Iterable values, .indexOf("noindex") != -1) { // Delete it! context.write(key, DELETE_ACTION); - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_DELETED_ROBOTS_NOINDEX_TOTAL).increment(1); + deletedRobotsNoIndexCounter.increment(1); return; } } @@ -317,8 +358,7 @@ public void reduce(Text key, Iterable values, if (delete && fetchDatum != null) { if (fetchDatum.getStatus() == CrawlDatum.STATUS_FETCH_GONE || dbDatum != null && dbDatum.getStatus() == CrawlDatum.STATUS_DB_GONE) { - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_DELETED_GONE_TOTAL).increment(1); + deletedGoneCounter.increment(1); context.write(key, DELETE_ACTION); return; } @@ -327,8 +367,7 @@ public void reduce(Text key, Iterable values, || fetchDatum.getStatus() == CrawlDatum.STATUS_FETCH_REDIR_TEMP || dbDatum != null && dbDatum.getStatus() == CrawlDatum.STATUS_DB_REDIR_PERM || dbDatum != null && dbDatum.getStatus() == CrawlDatum.STATUS_DB_REDIR_TEMP) { - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_DELETED_REDIRECTS_TOTAL).increment(1); + deletedRedirectsCounter.increment(1); context.write(key, DELETE_ACTION); return; } @@ -340,16 +379,14 @@ public void reduce(Text key, Iterable values, // Whether to delete pages marked as duplicates if (delete && dbDatum != null && dbDatum.getStatus() == CrawlDatum.STATUS_DB_DUPLICATE) { - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_DELETED_DUPLICATES_TOTAL).increment(1); + deletedDuplicatesCounter.increment(1); context.write(key, DELETE_ACTION); return; } // Whether to skip DB_NOTMODIFIED pages if (skip && dbDatum != null && dbDatum.getStatus() == CrawlDatum.STATUS_DB_NOTMODIFIED) { - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_SKIPPED_NOT_MODIFIED_TOTAL).increment(1); + skippedNotModifiedCounter.increment(1); return; } @@ -379,8 +416,7 @@ public void reduce(Text key, Iterable values, boost = scfilters.indexerScore(key, doc, dbDatum, fetchDatum, parse, inlinks, boost); } catch (final ScoringFilterException e) { - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_ERRORS_SCORING_FILTER_TOTAL).increment(1); + errorsScoringFilterCounter.increment(1); LOG.warn("Error calculating score {}: {}", key, e); return; } @@ -415,8 +451,7 @@ public void reduce(Text key, Iterable values, doc = filters.filter(doc, parse, key, fetchDatum, inlinks); } catch (final IndexingException e) { LOG.warn("Error indexing {}: ", key, e); - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_ERRORS_INDEXING_FILTER_TOTAL).increment(1); + errorsIndexingFilterCounter.increment(1); return; } @@ -426,11 +461,9 @@ public void reduce(Text key, Iterable values, if (deleteSkippedByIndexingFilter) { NutchIndexAction action = new NutchIndexAction(null, NutchIndexAction.DELETE); context.write(key, action); - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_DELETED_BY_INDEXING_FILTER_TOTAL).increment(1); + deletedByIndexingFilterCounter.increment(1); } else { - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_SKIPPED_BY_INDEXING_FILTER_TOTAL).increment(1); + skippedByIndexingFilterCounter.increment(1); } return; } @@ -453,8 +486,7 @@ public void reduce(Text key, Iterable values, // Record indexing latency indexLatencyTracker.record(System.currentTimeMillis() - indexStart); - context.getCounter(NutchMetrics.GROUP_INDEXER, - NutchMetrics.INDEXER_INDEXED_TOTAL).increment(1); + indexedCounter.increment(1); NutchIndexAction action = new NutchIndexAction(doc, NutchIndexAction.ADD); context.write(key, action); diff --git a/src/java/org/apache/nutch/tools/warc/WARCExporter.java b/src/java/org/apache/nutch/tools/warc/WARCExporter.java index df4f6af057..96e8c5a974 100644 --- a/src/java/org/apache/nutch/tools/warc/WARCExporter.java +++ b/src/java/org/apache/nutch/tools/warc/WARCExporter.java @@ -41,6 +41,7 @@ import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; +import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.Job; @@ -112,6 +113,31 @@ public static class WARCReducer // Metadata to JSON Gson gson = new Gson(); + // Cached counter references to avoid repeated lookups in hot paths + private Counter missingContentCounter; + private Counter missingMetadataCounter; + private Counter omittedEmptyResponseCounter; + private Counter invalidUriCounter; + private Counter recordsGeneratedCounter; + private Counter exceptionCounter; + + @Override + public void setup(Context context) { + // Initialize cached counter references + missingContentCounter = context.getCounter( + NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_MISSING_CONTENT_TOTAL); + missingMetadataCounter = context.getCounter( + NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_MISSING_METADATA_TOTAL); + omittedEmptyResponseCounter = context.getCounter( + NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_OMITTED_EMPTY_RESPONSE_TOTAL); + invalidUriCounter = context.getCounter( + NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_INVALID_URI_TOTAL); + recordsGeneratedCounter = context.getCounter( + NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_RECORDS_GENERATED_TOTAL); + exceptionCounter = context.getCounter( + NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_EXCEPTION_TOTAL); + } + @Override public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { @@ -148,15 +174,13 @@ public void reduce(Text key, Iterable values, // check that we have everything we need if (content == null) { LOG.info("Missing content for {}", key); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_MISSING_CONTENT_TOTAL).increment(1); + missingContentCounter.increment(1); return; } if (cd == null) { LOG.info("Missing fetch datum for {}", key); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_MISSING_METADATA_TOTAL).increment(1); + missingMetadataCounter.increment(1); return; } @@ -164,8 +188,7 @@ public void reduce(Text key, Iterable values, // Empty responses is everything that was not a regular response if (!(cd.getStatus() == CrawlDatum.STATUS_FETCH_SUCCESS || cd.getStatus() == CrawlDatum.STATUS_FETCH_NOTMODIFIED)) { - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_OMITTED_EMPTY_RESPONSE_TOTAL).increment(1); + omittedEmptyResponseCounter.increment(1); return; } } @@ -240,8 +263,7 @@ public void reduce(Text key, Iterable values, .append(uri.toASCIIString()).append(CRLF); } catch (Exception e) { LOG.error("Invalid URI {} ", key); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_INVALID_URI_TOTAL).increment(1); + invalidUriCounter.increment(1); return; } @@ -273,14 +295,12 @@ public void reduce(Text key, Iterable values, new ByteArrayInputStream(bos.toByteArray())); WARCRecord record = new WARCRecord(in); context.write(NullWritable.get(), new WARCWritable(record)); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_RECORDS_GENERATED_TOTAL).increment(1); + recordsGeneratedCounter.increment(1); } catch (IOException | IllegalStateException exception) { LOG.error( "Exception when generating WARC resource record for {} : {}", key, exception.getMessage()); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_EXCEPTION_TOTAL).increment(1); + exceptionCounter.increment(1); } // Do we need to emit a metadata record too? @@ -322,8 +342,7 @@ public void reduce(Text key, Iterable values, .append(uri.toASCIIString()).append(CRLF); } catch (Exception e) { LOG.error("Invalid URI {} ", key); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_INVALID_URI_TOTAL).increment(1); + invalidUriCounter.increment(1); return; } @@ -339,14 +358,12 @@ public void reduce(Text key, Iterable values, new ByteArrayInputStream(bos.toByteArray())); WARCRecord record = new WARCRecord(in); context.write(NullWritable.get(), new WARCWritable(record)); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_RECORDS_GENERATED_TOTAL).increment(1); + recordsGeneratedCounter.increment(1); } catch (IOException | IllegalStateException exception) { LOG.error( "Exception when generating WARC metadata record for {} : {}", key, exception.getMessage(), exception); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_EXCEPTION_TOTAL).increment(1); + exceptionCounter.increment(1); } } @@ -384,8 +401,7 @@ public void reduce(Text key, Iterable values, .append(uri.toASCIIString()).append(CRLF); } catch (Exception e) { LOG.error("Invalid URI {} ", key); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_INVALID_URI_TOTAL).increment(1); + invalidUriCounter.increment(1); return; } @@ -401,14 +417,12 @@ public void reduce(Text key, Iterable values, new ByteArrayInputStream(bos.toByteArray())); WARCRecord record = new WARCRecord(in); context.write(NullWritable.get(), new WARCWritable(record)); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_RECORDS_GENERATED_TOTAL).increment(1); + recordsGeneratedCounter.increment(1); } catch (IOException | IllegalStateException exception) { LOG.error( "Exception when generating WARC metadata record for {} : {}", key, exception.getMessage(), exception); - context.getCounter(NutchMetrics.GROUP_WARC_EXPORTER, - NutchMetrics.WARC_EXCEPTION_TOTAL).increment(1); + exceptionCounter.increment(1); } } } diff --git a/src/java/org/apache/nutch/util/SitemapProcessor.java b/src/java/org/apache/nutch/util/SitemapProcessor.java index 7055a6d86a..a0378ec63d 100644 --- a/src/java/org/apache/nutch/util/SitemapProcessor.java +++ b/src/java/org/apache/nutch/util/SitemapProcessor.java @@ -31,6 +31,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.Writable; +import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; @@ -114,6 +115,13 @@ private static class SitemapMapper extends Mapper values, Context context) originalDatum.setModifiedTime(sitemapDatum.getModifiedTime()); } - context.getCounter(NutchMetrics.GROUP_SITEMAP, - NutchMetrics.SITEMAP_EXISTING_ENTRIES_TOTAL).increment(1); + existingEntriesCounter.increment(1); context.write(key, originalDatum); } else if(sitemapDatum != null) { // For the newly discovered links via sitemap, set the status as unfetched and emit - context.getCounter(NutchMetrics.GROUP_SITEMAP, - NutchMetrics.SITEMAP_NEW_ENTRIES_TOTAL).increment(1); + newEntriesCounter.increment(1); sitemapDatum.setStatus(CrawlDatum.STATUS_DB_UNFETCHED); context.write(key, sitemapDatum); } @@ -465,11 +487,11 @@ public void sitemap(Path crawldb, Path hostdb, Path sitemapUrlDir, boolean stric FSUtils.replace(fs, current, tempCrawlDb, true); LockUtil.removeLockFile(fs, lock); - long filteredRecords = job.getCounters().findCounter("Sitemap", "filtered_records").getValue(); - long fromHostname = job.getCounters().findCounter("Sitemap", "sitemaps_from_hostname").getValue(); - long fromSeeds = job.getCounters().findCounter("Sitemap", "sitemap_seeds").getValue(); - long failedFetches = job.getCounters().findCounter("Sitemap", "failed_fetches").getValue(); - long newSitemapEntries = job.getCounters().findCounter("Sitemap", "new_sitemap_entries").getValue(); + long filteredRecords = job.getCounters().findCounter(NutchMetrics.GROUP_SITEMAP, NutchMetrics.SITEMAP_FILTERED_RECORDS_TOTAL).getValue(); + long fromHostname = job.getCounters().findCounter(NutchMetrics.GROUP_SITEMAP, NutchMetrics.SITEMAP_FROM_HOSTNAME_TOTAL).getValue(); + long fromSeeds = job.getCounters().findCounter(NutchMetrics.GROUP_SITEMAP, NutchMetrics.SITEMAP_SEEDS_TOTAL).getValue(); + long failedFetches = job.getCounters().findCounter(NutchMetrics.GROUP_SITEMAP, NutchMetrics.SITEMAP_FAILED_FETCHES_TOTAL).getValue(); + long newSitemapEntries = job.getCounters().findCounter(NutchMetrics.GROUP_SITEMAP, NutchMetrics.SITEMAP_NEW_ENTRIES_TOTAL).getValue(); LOG.info("SitemapProcessor: Total records rejected by filters: {}", filteredRecords); LOG.info("SitemapProcessor: Total sitemaps from host name: {}", fromHostname);