This PR refactors loadArchives() function to enable user to limit the size of individual record, preventing Spark jobs from failing due to large records.

GitHub issue(s):

If you are responding to an issue, please mention their numbers below.

• #247

What does this Pull Request do?

This PR adds the functionality to the loadArchives() function to limit size of individual records to be processed, preventing Spark jobs from failing due to records that are too big.

How should this be tested?

• git fetch --all
• git checkout limit-record-size
• mvn clean install
• Create an output directory with sub-directories:
  mkdir -p path/to/where/ever/you/can/write/output/all-domains path/to/where/ever/you/can/write/spark-jobs
• Adapt the script below:

import io.archivesunleashed._
import io.archivesunleashed.app._
import io.archivesunleashed.matchbox._
sc.setLogLevel("INFO")

//val input = "/tuna1/scratch/aut-issue-271/warcs/*.gz"
val input = "/tuna1/scratch/aut-issue-271/warcs/ARCHIVEIT-499-MONTHLY-JOB311846-20170703003948338-00020.warc.gz"

val output1 = "/tuna1/scratch/aut-issue-271/derivatives/all-domains"

val maxRecordSize: Option[Long] = Some(174552L)

RecordLoader.loadArchives(input, sc, maxRecordSize).keepValidPages().map(r => r.getContentBytes.length).sortBy(r => r, ascending = false).saveAsTextFile(output1)

sys.exit

• Run the following command with adapted paths with Apache Spark 2.1.3:
  /home/b25lin/spark-2.1.3-bin-hadoop2.7/bin/spark-shell --master local[10] --driver-memory 30G --conf spark.network.timeout=100000000 --conf spark.executor.heartbeatInterval=6000s --conf spark.driver.maxResultSize=10G --jars "/tuna1/scratch/borislin/aut/target/aut-0.17.1-SNAPSHOT-fatjar.jar" -i /tuna1/scratch/aut-issue-271/spark_jobs/499.scala | tee /tuna1/scratch/aut-issue-271/spark_jobs/499.scala.log

• Change the value of the maxRecordSize variable in the above script to a different value or remove it. See different results being produced. Only records that have sizes that are smaller or equal to the user-provided maxRecordSize will be filtered out.

Additional Notes:

• This PR doesn't affect our current use and documentation for the loadArchives() function since maxRecordSize is an optional parameter. But we should update our documentation to reflect that now user can use this maxRecordSize variable to limit record size.

Results

My results are in /tuna1/scratch/aut-issue-271/derivatives and log is in /tuna1/scratch/aut-issue-271/spark_jobs/499.scala.log

Interested parties

@lintool @ianmilligan1 @ruebot