Mvn2sbt

sparkler-core/conf/log4j.properties

@@ -30,6 +30,12 @@ appender.console.name = STDOUT
 appender.console.layout.type = PatternLayout
 appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
 
+rootLogger.level = warn
+rootLogger.appenderRefs = stdout
+rootLogger.appenderRef.stdout.ref = STDOUT
+logger.irds.name = edu.usc.irds
+logger.irds.level=DEBUG
+
 #rootLogger.level = INFO
 #rootLogger.appenderRefs = STDOUT
 #rootLogger.appenderRef.stdout.ref = STDOUT

sparkler-core/project/Dependencies.scala

@@ -73,6 +73,6 @@ object Dependencies {
     lazy val sql = group %% "spark-sql" % version % "provided"
   }
   lazy val tikaParsers = "org.apache.tika" % "tika-parsers" % "1.24"
-  lazy val elasticsearch = "org.elasticsearch.client" % "elasticsearch-rest-high-level-client" % "7.16.2"
+  lazy val elasticsearch = "org.elasticsearch.client" % "elasticsearch-rest-high-level-client" % "7.16.3"
 
 }

sparkler-core/sparkler-api/src/main/java/edu/usc/irds/sparkler/SparklerConfiguration.java

@@ -104,4 +104,14 @@ public String getDatabaseURI() {
         return (String) this.get(dbToUse+".uri");
     }
 
+    public String getDatabaseUsername(){
+        String dbToUse = (String) this.getOrDefault(Constants.key.CRAWLDB_BACKEND, "solr"); // solr is default
+        return (String) this.getOrDefault(dbToUse+".username", "");
+    }
+
+    public String getDatabasePassword(){
+        String dbToUse = (String) this.getOrDefault(Constants.key.CRAWLDB_BACKEND, "solr"); // solr is default
+        return (String) this.getOrDefault(dbToUse+".password", "");
+    }
+
 }

sparkler-core/sparkler-api/src/main/java/edu/usc/irds/sparkler/util/FetcherDefault.java

@@ -247,6 +247,8 @@ public FetchedData fetch(Resource resource) throws Exception {
         if (truncated) {
             fetchedData.getHeaders().put(TRUNCATED, Collections.singletonList(Boolean.TRUE.toString()));
         }
+
+        response2.close();
         return fetchedData;
 
     }

sparkler-core/sparkler-app/src/main/resources/log4j2.properties

@@ -41,21 +41,8 @@ appender.console.name = STDOUT
 appender.console.layout.type = PatternLayout
 appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
 
-rootLogger.level = INFO
-rootLogger.appenderRefs = STDOUT
+rootLogger.level = warn
+rootLogger.appenderRefs = stdout
 rootLogger.appenderRef.stdout.ref = STDOUT
 logger.irds.name = edu.usc.irds
 logger.irds.level=DEBUG
-logger.kythera.name = com.kytheralabs
-logger.kythera.level = DEBUG
-
-logger.pf4j.name = org.pf4j
-logger.pf4j.level = debug
-logger.pf4j.additivity = false
-logger.pf4j.appenderRef.console.ref = console
-#logger.loader.name = org.pf4j.PluginClassLoader
-#logger.loader.level = trace
-#logger.finder.name = org.pf4j.AbstractExtensionFinder
-#logger.finder.level = trace
-logger.spicule.name = uk.co.spicule
-logger.spicule.level = DEBUG

sparkler-core/sparkler-app/src/main/scala/edu/usc/irds/sparkler/pipeline/Crawler.scala

@@ -268,6 +268,7 @@ class Crawler extends CliTool {
     GenericFunction(job, GenericProcess.Event.SHUTDOWN,new SQLContext(sc).sparkSession, null)
     LOG.info("Shutting down Spark CTX..")
     sc.stop()
+    System.exit(0)
   }
 
   def scoreAndStore(fetchedRdd: RDD[CrawlData], taskId: String, storageProxy: StorageProxy, storageFactory: StorageProxyFactory): Unit ={

sparkler-core/sparkler-app/src/main/scala/edu/usc/irds/sparkler/storage/elasticsearch/ElasticsearchProxy.scala

@@ -22,50 +22,73 @@ import edu.usc.irds.sparkler.base.Loggable
 import edu.usc.irds.sparkler.model.Resource
 import edu.usc.irds.sparkler.storage.StorageProxy
 import org.apache.http.HttpHost
+import org.apache.http.auth.{AuthScope, UsernamePasswordCredentials}
+import org.apache.http.impl.client.BasicCredentialsProvider
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder
 import org.elasticsearch.action.index.{IndexRequest, IndexResponse}
 import org.elasticsearch.action.update.UpdateRequest
-import org.elasticsearch.client.{RequestOptions, RestClient, RestHighLevelClient}
+import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
 import org.elasticsearch.script.Script
 import org.elasticsearch.xcontent.{XContentBuilder, XContentFactory}
 
 import java.io.{Closeable, IOException}
 import java.util.AbstractMap.SimpleEntry
 import scala.collection.mutable.ArrayBuffer
 
-
 /**
   *
   * @since 3/6/21
   */
 class ElasticsearchProxy(var config: SparklerConfiguration) extends StorageProxy with Closeable with Loggable {
 
-  // creates the client
-  private var crawlDb = newClient(config.getDatabaseURI)
+  var conn : Option[RestHighLevelClient] = None
 
   private var indexRequests = ArrayBuffer[IndexRequest]()
 
   def newClient(crawlDbUri: String): RestHighLevelClient = {
-    val scheme : String = crawlDbUri.substring(0, crawlDbUri.indexOf(':'))
-    val hostname : String = crawlDbUri.substring(crawlDbUri.indexOf(':') + 3, crawlDbUri.lastIndexOf(':'))
-    val port : Int = Integer.valueOf(crawlDbUri.substring(crawlDbUri.lastIndexOf(':') + 1))
-
-    if (scheme.equals("http") || scheme.equals("https")) {
-      new RestHighLevelClient(
-        RestClient.builder(
-          new HttpHost(hostname, port, scheme)
-        )
-      )
-    } else if (crawlDbUri.startsWith("file://")) {
-      ???  // TODO: embedded ES?
-    } else if (crawlDbUri.contains("::")){
-      ???  // TODO: cloudmode with zookeepers ES?
-    } else {
-      throw new RuntimeException(s"$crawlDbUri not supported")
+    if(conn.isEmpty) {
+
+
+      val scheme = crawlDbUri.substring(0, crawlDbUri.indexOf(':'))
+      val hostname: String = crawlDbUri.substring(crawlDbUri.indexOf(':') + 3, crawlDbUri.lastIndexOf(':'))
+      val port: Int = Integer.valueOf(crawlDbUri.substring(crawlDbUri.lastIndexOf(':') + 1))
+
+      if (scheme.equals("http") || scheme.equals("https")) {
+        if(config.getDatabaseUsername.nonEmpty) {
+          val credentialsProvider = new BasicCredentialsProvider
+          credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(config.getDatabaseUsername, config.getDatabasePassword))
+          conn = Some(new RestHighLevelClient(
+            RestClient.builder(
+              new HttpHost(hostname, port, scheme)
+            ).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
+              override def customizeHttpClient(httpClientBuilder: HttpAsyncClientBuilder): HttpAsyncClientBuilder =
+                httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
+            })
+          ))
+        } else{
+          conn = Some(new RestHighLevelClient(
+            RestClient.builder(
+              new HttpHost(hostname, port, scheme)
+            )))
+        }
+      } else if (crawlDbUri.startsWith("file://")) {
+        ??? // TODO: embedded ES?
+      } else if (crawlDbUri.contains("::")) {
+        ??? // TODO: cloudmode with zookeepers ES?
+      } else {
+        throw new RuntimeException(s"$crawlDbUri not supported")
+      }
+      conn.get
+    } else{
+      conn.get
     }
   }
 
   def getClient(): RestHighLevelClient = {
-    crawlDb
+    if(conn.isEmpty){
+      conn = Some(newClient(config.getDatabaseURI))
+    }
+    conn.get
   }
 
   def addResourceDocs(docs: java.util.Iterator[Map[String, Object]]): Unit = {
@@ -119,7 +142,7 @@ class ElasticsearchProxy(var config: SparklerConfiguration) extends StorageProxy
               val newScript: Script = new Script(scriptCode)
               updateRequestForScripts.script(newScript)
 
-              crawlDb.update(updateRequestForScripts, RequestOptions.DEFAULT)
+              getClient().update(updateRequestForScripts, RequestOptions.DEFAULT)
               updateRequestForScripts.retryOnConflict(3)
             }
             else {
@@ -137,7 +160,7 @@ class ElasticsearchProxy(var config: SparklerConfiguration) extends StorageProxy
         .upsert(indexRequest) // upsert either updates or insert if not found
 
       updateRequest.retryOnConflict(3)
-      crawlDb.update(updateRequest, RequestOptions.DEFAULT)
+      getClient().update(updateRequest, RequestOptions.DEFAULT)
     }
     catch {
       case e: IOException =>
@@ -149,7 +172,7 @@ class ElasticsearchProxy(var config: SparklerConfiguration) extends StorageProxy
     for (indexRequest <- indexRequests) {
       var response : IndexResponse = null
       try {
-        response = crawlDb.index(indexRequest, RequestOptions.DEFAULT)
+        response = getClient().index(indexRequest, RequestOptions.DEFAULT)
       }
       catch {
         case e: IOException =>
@@ -161,7 +184,7 @@ class ElasticsearchProxy(var config: SparklerConfiguration) extends StorageProxy
 
   def close(): Unit = {
     commitCrawlDb() // make sure buffer is flushed
-    crawlDb.close()
+    getClient().close()
   }
 
 }

sparkler-core/sparkler-app/src/main/scala/edu/usc/irds/sparkler/storage/elasticsearch/ElasticsearchRDD.scala

@@ -14,13 +14,14 @@ import org.elasticsearch.index.query.BoolQueryBuilder
 import org.elasticsearch.client.RestHighLevelClient
 import org.elasticsearch.search.sort.SortOrder
 import org.elasticsearch.search.aggregations.AggregationBuilders
-import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder
+import org.elasticsearch.search.aggregations.bucket.terms.{ParsedTerms, TermsAggregationBuilder}
 import org.elasticsearch.search.aggregations.Aggregations
 import org.elasticsearch.search.aggregations.Aggregation
 import org.apache.lucene.queryparser.classic.QueryParserBase
 import org.elasticsearch.search.SearchHits
 import org.elasticsearch.search.SearchHit
 import org.elasticsearch.common.document.DocumentField
+import org.elasticsearch.search.aggregations.bucket.histogram.Histogram
 
 import scala.collection.JavaConversions._
 
@@ -136,7 +137,7 @@ class ElasticsearchRDD(sc: SparkContext,
     }
 
     // grouping
-    var groupBy : TermsAggregationBuilder = AggregationBuilders.terms("by" + Constants.storage.PARENT)
+    val groupBy : TermsAggregationBuilder = AggregationBuilders.terms("by" + Constants.storage.PARENT)
                                                           .field(Constants.storage.PARENT + ".keyword")
     groupBy.size(1)
     searchSourceBuilder.aggregation(groupBy)
@@ -153,12 +154,15 @@ class ElasticsearchRDD(sc: SparkContext,
     }
 
     val searchResponse : SearchResponse = client.search(searchRequest, RequestOptions.DEFAULT)
-    val shs : SearchHits = searchResponse.getHits
-    val res = new Array[Partition](shs.getTotalHits.value.toInt)
-    for (i <- 0 until shs.getTotalHits.value.toInt) {
-      //TODO: improve partitioning : (1) club smaller domains, (2) support for multiple partitions for larger domains
-      res(i) = new SparklerGroupPartition(i, shs.getHits()(i).getSourceAsMap.get(Constants.storage.PARENT).asInstanceOf[String])
-    }
+    val aggmap = searchResponse.getAggregations.getAsMap
+    val agg2 = aggmap.head._2.asInstanceOf[ParsedTerms]
+    val res = new Array[Partition](agg2.getBuckets.size())
+
+    var i = 0
+    agg2.getBuckets.foreach(b => {
+      res(i) = new SparklerGroupPartition(i, b.getKeyAsString)
+      i = i + 1
+    })
 
     proxy.close()
     res