Add Data Quality Monitoring for Extractors

.gitignore

@@ -7,3 +7,4 @@ target/
 *.tmp
 java_pid*
 dump/test-basedir
+.settings

core/.project

@@ -1,13 +1,35 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <projectDescription>
-    <name>core</name>
-    <buildSpec>
-        <buildCommand>
-            <name>org.scala-ide.sdt.core.scalabuilder</name>
-        </buildCommand>
-    </buildSpec>
-    <natures>
-        <nature>org.scala-ide.sdt.core.scalanature</nature>
-        <nature>org.eclipse.jdt.core.javanature</nature>
-    </natures>
-</projectDescription>
+	<name>core</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.scala-ide.sdt.core.scalabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+		<nature>org.scala-ide.sdt.core.scalanature</nature>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+	</natures>
+	<filteredResources>
+		<filter>
+			<id>1765562241624</id>
+			<name></name>
+			<type>30</type>
+			<matcher>
+				<id>org.eclipse.core.resources.regexFilterMatcher</id>
+				<arguments>node_modules|\.git|__CREATED_BY_JAVA_LANGUAGE_SERVER__</arguments>
+			</matcher>
+		</filter>
+	</filteredResources>
+</projectDescription>

core/src/main/scala/org/dbpedia/extraction/mappings/HomepageExtractor.scala

@@ -6,7 +6,7 @@ import org.dbpedia.extraction.transform.Quad
 import org.dbpedia.extraction.wikiparser._
 import org.dbpedia.extraction.config.mappings.HomepageExtractorConfig
 import org.dbpedia.extraction.ontology.Ontology
-import org.dbpedia.extraction.util.Language
+import org.dbpedia.extraction.util.{Language, DataQualityMonitor}
 import org.dbpedia.iri.{IRISyntaxException, UriUtils}
 
 import scala.language.reflectiveCalls
@@ -26,6 +26,9 @@ extends PageNodeExtractor
 {
   private val language = context.language.wikiCode
 
+  // Extraction quality monitor for logging and metrics
+  private val monitor = DataQualityMonitor.forExtractor("HomepageExtractor")
+
   private val propertyNames = HomepageExtractorConfig.propertyNames(language)
 
   private val official = HomepageExtractorConfig.official(language)
@@ -48,7 +51,10 @@ extends PageNodeExtractor
 
   override def extract(page: PageNode, subjectUri: String): Seq[Quad] =
   {
-    if(page.title.namespace != Namespace.Main) return Seq.empty
+    if(page.title.namespace != Namespace.Main) {
+      monitor.logSkipped(page.title.encoded, s"Not in main namespace: ${page.title.namespace}")
+      return Seq.empty
+    }
 
     val list = collectProperties(page).filter(p => propertyNames.contains(p.key.toLowerCase)).flatMap {
       NodeUtil.splitPropertyNode(_, splitPropertyNodeLinkStrict, true)
@@ -118,12 +124,34 @@ extends PageNodeExtractor
   {
     UriUtils.createURI(url) match{
       case Success(u) => UriUtils.cleanLink(u) match{
-        case Some(c) => Seq(new Quad(context.language, DBpediaDatasets.Homepages, subjectUri, homepageProperty, c , node.sourceIri))
-        case None => Seq()
+        case Some(c) =>
+          monitor.logSuccess(subjectUri, 1)
+          Seq(new Quad(context.language, DBpediaDatasets.Homepages, subjectUri, homepageProperty, c , node.sourceIri))
+        case None =>
+          monitor.logInvalidData(
+            subjectUri,
+            "URL could not be cleaned",
+            data = Some(url)
+          )
+          Seq()
       }
       case Failure(f) => f match{
-        case _ : IRISyntaxException => Seq()  //   TODO: log
-        case _ => Seq()
+        case ex: IRISyntaxException =>
+          monitor.logInvalidData(
+            subjectUri,
+            "Malformed IRI syntax",
+            exception = Some(ex),
+            data = Some(url)
+          )
+          Seq()
+        case ex =>
+          monitor.logInvalidData(
+            subjectUri,
+            "Unexpected error creating URI",
+            exception = Some(ex),
+            data = Some(url)
+          )
+          Seq()
       }
     }
   }

core/src/main/scala/org/dbpedia/extraction/util/DataQualityMonitor.scala

@@ -0,0 +1,199 @@
+package org.dbpedia.extraction.util
+
+import java.util.logging.{Level, Logger}
+import java.util.concurrent.atomic.AtomicLong
+import scala.collection.concurrent.TrieMap
+
+/**
+ * Centralized monitoring and logging system for extraction quality and errors.
+ *
+ * Features:
+ * - Structured logging with context (extractor, page, error type)
+ * - Metrics collection for error rates
+ * - Thread-safe error counting
+ * - Export capabilities for failed extractions
+ *
+ * Usage:
+ * {{{
+ *   val monitor = DataQualityMonitor.forExtractor("HomepageExtractor")
+ *   monitor.logInvalidData("Einstein", "Invalid IRI: malformed URL", exception)
+ *   monitor.getMetrics() // Get error statistics
+ * }}}
+ */
+object DataQualityMonitor {
+
+  private val logger = Logger.getLogger(classOf[DataQualityMonitor].getName)
+
+  // Global metrics storage (thread-safe)
+  private val errorCounts = new TrieMap[String, AtomicLong]()
+  private val errorDetails = new TrieMap[String, collection.mutable.ListBuffer[ExtractionError]]()
+
+  /**
+   * Create a monitor for a specific extractor
+   */
+  def forExtractor(extractorName: String): DataQualityMonitor = {
+    new DataQualityMonitor(extractorName)
+  }
+
+  /**
+   * Get global extraction statistics
+   */
+  def getGlobalMetrics(): Map[String, Long] = {
+    errorCounts.map { case (key, counter) => (key, counter.get()) }.toMap
+  }
+
+  /**
+   * Get detailed errors for analysis
+   */
+  def getErrorDetails(errorType: String, limit: Int = 100): List[ExtractionError] = {
+    errorDetails.get(errorType) match {
+      case Some(errors) => errors.take(limit).toList
+      case None => List.empty
+    }
+  }
+
+  /**
+   * Export errors to CSV format for analysis
+   */
+  def exportToCsv(errorType: String, limit: Int = 1000): String = {
+    val errors = getErrorDetails(errorType, limit)
+    val header = "Extractor,PageTitle,ErrorMessage,Timestamp\n"
+    val rows = errors.map(e =>
+      s"${e.extractorName},${e.pageTitle},${e.message.replaceAll(",", ";")},${e.timestamp}"
+    ).mkString("\n")
+    header + rows
+  }
+
+  /**
+   * Reset all metrics (useful for testing)
+   */
+  def reset(): Unit = {
+    errorCounts.clear()
+    errorDetails.clear()
+  }
+}
+
+/**
+ * Monitor instance for a specific extractor
+ */
+class DataQualityMonitor(val extractorName: String) {
+
+  private val logger = Logger.getLogger(s"org.dbpedia.extraction.monitor.$extractorName")
+
+  /**
+   * Log invalid data with context
+   *
+   * @param pageTitle The Wikipedia page being processed
+   * @param reason Description of why the data is invalid
+   * @param exception Optional exception that caused the error
+   * @param data Optional invalid data for debugging
+   */
+  def logInvalidData(
+    pageTitle: String,
+    reason: String,
+    exception: Option[Throwable] = None,
+    data: Option[String] = None
+  ): Unit = {
+    val message = buildMessage(pageTitle, reason, data)
+
+    // Log with appropriate level
+    exception match {
+      case Some(ex) => logger.log(Level.WARNING, message, ex)
+      case None => logger.warning(message)
+    }
+
+    // Record metrics
+    recordError(pageTitle, reason, exception)
+  }
+
+  /**
+   * Log skipped extraction with reason
+   */
+  def logSkipped(pageTitle: String, reason: String): Unit = {
+    logger.fine(s"[$extractorName] Skipped '$pageTitle': $reason")
+  }
+
+  /**
+   * Log successful extraction with statistics
+   */
+  def logSuccess(pageTitle: String, triplesCount: Int): Unit = {
+    logger.fine(s"[$extractorName] Extracted $triplesCount triples from '$pageTitle'")
+  }
+
+  /**
+   * Get metrics for this extractor
+   */
+  def getMetrics(): Map[String, Long] = {
+    DataQualityMonitor.errorCounts
+      .filter { case (key, _) => key.startsWith(s"$extractorName:") }
+      .map { case (key, counter) => (key, counter.get()) }
+      .toMap
+  }
+
+  /**
+   * Get total error count for this extractor
+   */
+  def getTotalErrors(): Long = {
+    getMetrics().values.sum
+  }
+
+  // Private helper methods
+
+  private def buildMessage(pageTitle: String, reason: String, data: Option[String]): String = {
+    val dataStr = data.map(d => s" | Data: ${truncate(d, 200)}").getOrElse("")
+    s"[$extractorName] Invalid data in '$pageTitle': $reason$dataStr"
+  }
+
+  private def recordError(pageTitle: String, reason: String, exception: Option[Throwable]): Unit = {
+    val errorType = s"$extractorName:${categorizeError(reason, exception)}"
+
+    // Increment counter
+    DataQualityMonitor.errorCounts
+      .getOrElseUpdate(errorType, new AtomicLong(0))
+      .incrementAndGet()
+
+    // Store details (limit to prevent memory issues)
+    val errorDetail = ExtractionError(
+      extractorName = extractorName,
+      pageTitle = pageTitle,
+      message = reason,
+      exceptionType = exception.map(_.getClass.getSimpleName),
+      timestamp = System.currentTimeMillis()
+    )
+
+    DataQualityMonitor.errorDetails
+      .getOrElseUpdate(errorType, collection.mutable.ListBuffer.empty)
+      .synchronized {
+        val buffer = DataQualityMonitor.errorDetails(errorType)
+        if (buffer.size < 10000) { // Limit storage
+          buffer += errorDetail
+        }
+      }
+  }
+
+  private def categorizeError(reason: String, exception: Option[Throwable]): String = {
+    exception match {
+      case Some(ex) => ex.getClass.getSimpleName
+      case None if reason.toLowerCase.contains("invalid") => "InvalidData"
+      case None if reason.toLowerCase.contains("malformed") => "MalformedData"
+      case None if reason.toLowerCase.contains("missing") => "MissingData"
+      case None => "Other"
+    }
+  }
+
+  private def truncate(str: String, maxLength: Int): String = {
+    if (str.length <= maxLength) str
+    else str.substring(0, maxLength) + "..."
+  }
+}
+
+/**
+ * Case class representing an extraction error
+ */
+case class ExtractionError(
+  extractorName: String,
+  pageTitle: String,
+  message: String,
+  exceptionType: Option[String],
+  timestamp: Long
+)