Uri prediction

.travis.yml

@@ -2,8 +2,20 @@ sudo: required
 
 language: java
 
+jdk: oraclejdk8
+
+dist: trusty
+
+addons:
+  apt:
+    packages:
+      - oracle-java8-installer
+
+
+
 services:
   - docker
 
 before_install:
 - docker pull rethinkdb:2.3.5
+

docker-compose-sparql.yml

@@ -31,6 +31,7 @@ services:
       - ./data/frontier:/var/squirrel/data
       - ./seed/seeds.csv:/var/squirrel/seeds.csv:ro
       - ./whitelist/whitelist.txt:/var/squirrel/whitelist.txt:ro
+      - ./spring-config:/var/squirrel/spring-config
     command: java -cp squirrel.jar org.dice_research.squirrel.components.FrontierComponentStarter
 
   virtuosohost:
@@ -136,18 +137,18 @@ services:
         - ./spring-config:/var/squirrel/spring-config
     command: java -cp squirrel.jar org.dice_research.squirrel.components.WorkerComponentStarter
 
-  deduplicator:
-    image: squirrel
-    container_name: deduplicator
-    environment:
-      DEDUPLICATION_ACTIVE: "true"
-      HOBBIT_RABBIT_HOST: rabbit
-      OUTPUT_FOLDER: /var/squirrel/data
-      MDB_HOST_NAME: mongodb
-      MDB_PORT: 27017
-      SPARQL_HOST_NAME: sparqlhost
-      SPARQL_HOST_PORT: 3030
-      SERVICE_PRECONDITION: "rethinkdb:28015 rabbit:5672"
-    volumes:
-        - ./data/deduplicator:/var/squirrel/data
-    command: java -cp squirrel.jar org.hobbit.core.run.ComponentStarter org.aksw.simba.squirrel.components.DeduplicatorComponent
+#  deduplicator:
+#    image: squirrel
+#    container_name: deduplicator
+#    environment:
+#      DEDUPLICATION_ACTIVE: "true"
+#      HOBBIT_RABBIT_HOST: rabbit
+#      OUTPUT_FOLDER: /var/squirrel/data
+#      MDB_HOST_NAME: mongodb
+#      MDB_PORT: 27017
+#      SPARQL_HOST_NAME: sparqlhost
+#      SPARQL_HOST_PORT: 3030
+#      SERVICE_PRECONDITION: "rethinkdb:28015 rabbit:5672"
+#    volumes:
+#        - ./data/deduplicator:/var/squirrel/data
+#    command: java -cp squirrel.jar org.hobbit.core.run.ComponentStarter org.aksw.simba.squirrel.components.DeduplicatorComponent

spring-config/frontier-context.xml

@@ -54,4 +54,5 @@
 
 
 
+
 </beans>

squirrel.api/src/main/java/org/dice_research/squirrel/Constants.java

@@ -46,6 +46,32 @@ public class Constants {
      */
     public static final String URI_PREFERRED_RECRAWL_ON = "recrawl-on";
 
+    /*
+     * The data related to the predictor
+     */
+    /**
+     * This key stores the value predicted by the Predictor for each URI denoting
+     * the class it belongs to (Positive class or Negative class)
+     */
+    public static final String URI_PREDICTED_LABEL = "predicted-label";
+    /**
+     * This key stores the value denoting the true class of each URI
+     */
+    public static final String URI_TRUE_LABEL = "true-label";
+    /**
+     * This key stores an integer denoting the true class of the URI
+     */
+    public static final String URI_TRUE_CLASS = "true_class";
+    /**
+     * This key stores the feature vector generated for each URI for prediction purpose
+     */
+    public static final String FEATURE_VECTOR = "feature-vector";
+    /**
+     * This key stores the parent URI of each crawled URI
+     */
+    public static final String REFERRING_URI = "referring-uri";
+
+
     //////////////////////////////////////////////////
     // URIs
     //////////////////////////////////////////////////

squirrel.frontier/pom.xml

@@ -21,6 +21,12 @@
 			<groupId>org.dice-research</groupId>
 			<artifactId>squirrel.web-api</artifactId>
 		</dependency>
+		<!-- Online machine learning library used to implement better URI type prediction in crawler -->
+		<dependency>
+			<groupId>de.jungblut.ml</groupId>
+			<artifactId>tjungblut-online-ml</artifactId>
+			<version>0.5</version>
+		</dependency>
 
 		<!-- ~~~~~~~~~~~~~~~~~~~ Testing ~~~~~~~~~~~~~~~~~~~~~~ -->
 		<!-- JUnit -->
@@ -63,4 +69,4 @@
 			</plugin>
 		</plugins>
 	</build>
-</project>
+</project>

squirrel.frontier/src/main/java/org/dice_research/squirrel/components/FrontierComponent.java

@@ -35,7 +35,9 @@
 import org.dice_research.squirrel.frontier.impl.QueueBasedTerminationCheck;
 import org.dice_research.squirrel.frontier.impl.TerminationCheck;
 import org.dice_research.squirrel.frontier.impl.WorkerGuard;
+import org.dice_research.squirrel.predictor.*;
 import org.dice_research.squirrel.queue.InMemoryQueue;
+import org.dice_research.squirrel.queue.IpAddressBasedQueue;
 import org.dice_research.squirrel.queue.UriQueue;
 import org.dice_research.squirrel.rabbit.RPCServer;
 import org.dice_research.squirrel.rabbit.RespondingDataHandler;
@@ -54,6 +56,7 @@
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 
+
 @Component
 @Qualifier("frontierComponent")
 public class FrontierComponent extends AbstractComponent implements RespondingDataHandler {
@@ -77,10 +80,16 @@ public class FrontierComponent extends AbstractComponent implements RespondingDa
     private final WorkerGuard workerGuard = new WorkerGuard(this);
     private final boolean doRecrawling = true;
     private long recrawlingTime = 1000L * 60L * 60L * 24L * 30;
+
+
     private Timer timerTerminator;
 
+
     public static final boolean RECRAWLING_ACTIVE = true;
 
+
+    protected Predictor predictor;
+
     @Override
     public void init() throws Exception {
         super.init();
@@ -108,9 +117,15 @@ public void init() throws Exception {
             queue = new InMemoryQueue();
             knownUriFilter = new InMemoryKnownUriFilter(doRecrawling, recrawlingTime);
         }
+        // Training the URI predictor model with a training dataset
+        try {
+            predictor = new MultinomialPredictor.MultinomialPredictorBuilder().withFile("multiNomialTrainData.txt").build();
+        }catch (Exception e){
+            e.printStackTrace();
+        }
 
         // Build frontier
-        frontier = new ExtendedFrontierImpl(new NormalizerImpl(), knownUriFilter, uriReferences, queue, doRecrawling);
+        frontier = new ExtendedFrontierImpl(new NormalizerImpl(), knownUriFilter, uriReferences, (IpAddressBasedQueue) queue, doRecrawling, predictor);
 
         rabbitQueue = this.incomingDataQueueFactory.createDefaultRabbitQueue(Constants.FRONTIER_QUEUE_NAME);
         receiver = (new RPCServer.Builder()).responseQueueFactory(outgoingDataQueuefactory).dataHandler(this)
@@ -139,11 +154,13 @@ public void init() throws Exception {
                     + webConfiguration.isVisualizationOfCrawledGraphEnabled()
                     + ". No WebServiceSenderThread will be started!");
         }
+
+
     }
 
     @Override
     public void run() throws Exception {
-        
+
         terminationMutex.acquire();
     }
 
@@ -177,7 +194,7 @@ public void handleData(byte[] data) {
 
     @Override
     public void handleData(byte[] data, ResponseHandler handler, String responseQueueName, String correlId) {
-    	
+
         Object deserializedData;
         try {
             deserializedData = serializer.deserialize(data);
@@ -200,7 +217,7 @@ public void handleData(byte[] data, ResponseHandler handler, String responseQueu
             if (deserializedData instanceof UriSetRequest) {
                 responseToUriSetRequest(handler, responseQueueName, correlId, (UriSetRequest) deserializedData);
             } else if (deserializedData instanceof UriSet) {
-            	
+
             	if(timerTerminator == null) {
             		LOGGER.info("Initializing Terminator task...");
                 	TimerTask terminatorTask = new TerminatorTask(queue, terminationMutex, this.workerGuard);
@@ -212,6 +229,7 @@ public void handleData(byte[] data, ResponseHandler handler, String responseQueu
             } else if (deserializedData instanceof CrawlingResult) {
                 CrawlingResult crawlingResult = (CrawlingResult) deserializedData;
                 LOGGER.warn("Received the message that the crawling for {} URIs is done.", crawlingResult.uris.size());
+
                 frontier.crawlingDone(crawlingResult.uris);
                 workerGuard.removeUrisForWorker(crawlingResult.idOfWorker, crawlingResult.uris);
             } else if (deserializedData instanceof AliveMessage) {
@@ -298,11 +316,11 @@ public void run() {
                     break;
                 }
             }
-            
+
 			if(!stillHasUris && terminationCheck.shouldFrontierTerminate(queue)) {
 	        	terminationMutex.release();
-	        }			
+	        }
         }
 
     }
-}
+}

squirrel.frontier/src/main/java/org/dice_research/squirrel/frontier/impl/ExtendedFrontierImpl.java

@@ -12,6 +12,7 @@
 import org.dice_research.squirrel.data.uri.norm.UriNormalizer;
 import org.dice_research.squirrel.deduplication.hashing.UriHashCustodian;
 import org.dice_research.squirrel.frontier.ExtendedFrontier;
+import org.dice_research.squirrel.predictor.Predictor;
 import org.dice_research.squirrel.queue.IpAddressBasedQueue;
 import org.dice_research.squirrel.queue.UriQueue;
 
@@ -29,11 +30,12 @@ public class ExtendedFrontierImpl extends FrontierImpl implements ExtendedFronti
      * @param generalRecrawlTime used to select the general Time after URIs should be recrawled. If Value is null the default Time is used.
      * @param timerPeriod        used to select if URIs should be recrawled.
      * @param uriHashCustodian   used to access and write hash values for uris.
+     * @param predictor          {@link Predictor}Used to predict the type of the URI
      */
     @SuppressWarnings("unused")
     public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFilter, UriQueue queue, boolean doesRecrawling,
-                                long generalRecrawlTime, long timerPeriod, UriHashCustodian uriHashCustodian) {
-        super(normalizer, knownUriFilter, queue, doesRecrawling, generalRecrawlTime, timerPeriod, uriHashCustodian);
+                                long generalRecrawlTime, long timerPeriod, UriHashCustodian uriHashCustodian, Predictor predictor) {
+        super(normalizer, knownUriFilter, queue, doesRecrawling, generalRecrawlTime, timerPeriod, uriHashCustodian, predictor);
     }
 
     /**
@@ -45,9 +47,10 @@ public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFil
      * @param queue            {@link UriQueue} used to manage the URIs that should be
      *                         crawled.
      * @param doesRecrawling   used to select if URIs should be recrawled.
+     * @param predictor         {@link Predictor}Used to predict the type of the URI
      */
-    public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFilter, IpAddressBasedQueue queue, boolean doesRecrawling) {
-        super(normalizer, knownUriFilter, queue, doesRecrawling);
+    public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFilter, IpAddressBasedQueue queue, boolean doesRecrawling, Predictor predictor) {
+        super(normalizer, knownUriFilter, queue, doesRecrawling, predictor);
     }
 
     /**
@@ -60,9 +63,12 @@ public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFil
      * @param queue          {@link UriQueue} used to manage the URIs that should be
      *                       crawled.
      * @param doesRecrawling used to select if URIs should be recrawled.
+     * @param predictor      PredictorImpl object used for prediction
      */
-    public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFilter, URIReferences uriReferences, UriQueue queue, boolean doesRecrawling) {
-        super(normalizer, knownUriFilter, uriReferences, queue, doesRecrawling);
+
+    public ExtendedFrontierImpl(UriNormalizer normalizer, KnownUriFilter knownUriFilter, URIReferences uriReferences, IpAddressBasedQueue queue, boolean doesRecrawling, Predictor predictor) {
+        super(normalizer, knownUriFilter, uriReferences, queue, doesRecrawling, predictor);
+
     }
 
     @Override
@@ -78,4 +84,4 @@ public void informAboutDeadWorker(String idOfWorker, List<CrawleableUri> lstUris
             setIps.forEach(ip -> ipQueue.markIpAddressAsAccessible(ip));
         }
     }
-}
+}