<commit_before>package vtpassim

import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions.{coalesce, concat, lit}

import collection.JavaConversions._
import scala.collection.mutable.{ArrayBuffer, StringBuilder}
import scala.util.Try

object APS {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("APS Application")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive", "true")

    val series = sqlContext.read.json(args(1))

    sc.binaryFiles(args(0), sc.defaultParallelism)
      .filter(_._1.endsWith(".zip"))
      .flatMap { x =>
        try {
          val fname = new java.io.File(new java.net.URL(x._1).toURI)
          val issue = fname.getName.replaceAll(".zip$", "")

          val zfile = new java.util.zip.ZipFile(fname)

          zfile.entries
            .filter { _.getName.endsWith(".xml") }
            .map { f =>
            val t = scala.xml.XML.load(zfile.getInputStream(f))
            val series = (t \ "Publication" \ "PublicationID").text
            val sdate = (t \ "NumericPubDate").text
            val date = Seq(sdate.substring(0, 4), sdate.substring(4, 6), sdate.substring(6, 8))
              .mkString("-")
            ("aps/" + (t \ "RecordID").text,
              "aps/" + series + "/" + date,
              series,
              date,
              (t \ "FullText").text.replaceAll("&", "&amp;").replaceAll("<", "&lt;")
                .replaceAll("([ ]{4,})", "\n$1"))
          }
        } catch {
          case e: Exception =>
            println(x._1 + e.toString)
            None
        }
      }
      .toDF("id", "issue", "apsseries", "date", "text")
      .join(series, Seq("apsseries"), "left_outer")
      .withColumn("series", coalesce('series, concat(lit("aps/"), 'apsseries)))
      .write.save(args(2))
  }
}<commit_msg>Skip empty records; add more metadata fields; filter on series dates where present.<commit_after>package vtpassim

import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions.{coalesce, concat, lit}

import collection.JavaConversions._
import scala.collection.mutable.{ArrayBuffer, StringBuilder}
import scala.util.Try

object APS {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("APS Application")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive", "true")

    val series = sqlContext.read.json(args(1))

    sc.binaryFiles(args(0), sc.defaultParallelism)
      .filter(_._1.endsWith(".zip"))
      .flatMap { x =>
        try {
          val fname = new java.io.File(new java.net.URL(x._1).toURI)
          val issue = fname.getName.replaceAll(".zip$", "")

          val zfile = new java.util.zip.ZipFile(fname)

          zfile.entries
            .filter { _.getName.endsWith(".xml") }
            .flatMap { f =>
            val t = scala.xml.XML.load(zfile.getInputStream(f))
            val series = (t \ "Publication" \ "PublicationID").text
            val sdate = (t \ "NumericPubDate").text
            val text = (t \ "FullText").text.replaceAll("&", "&amp;").replaceAll("<", "&lt;")
              .replaceAll("([ ]{4,})", "\n$1")

            if ( text != "" && series != "" && sdate != "" ) {
              val date = Seq(sdate.substring(0, 4), sdate.substring(4, 6), sdate.substring(6, 8))
                .mkString("-")
              Some(("aps/" + (t \ "RecordID").text,
                "aps/" + series + "/" + date,
                series, date, text,
                (t \ "RecordTitle").text,
                (t \ "ObjectType").text,
                (t \ "URLFullText").text,
                (t \ "LanguageCode").text.toLowerCase,
                (t \ "Contributor" \ "OriginalForm").text))
            } else
              None
          }
        } catch {
          case e: Exception =>
            println(x._1 + e.toString)
            None
        }
      }
      .toDF("id", "issue", "apsseries", "date", "text",
        "heading", "category", "url", "lang", "contributor")
      .join(series, Seq("apsseries"), "left_outer")
      .filter { ('startdate.isNull || 'startdate <= 'date) && ('enddate.isNull || 'enddate >= 'date) }
      .withColumn("series", coalesce('series, concat(lit("aps/"), 'apsseries)))
      .drop("apsseries", "startdate", "enddate", "title")
      .write.save(args(2))
  }
}
{'fraction_non_alphanumeric': 0.1657776829528483, 'fraction_numerical': 0.005973970556859398, 'mean_word_length': 2.2487872487872487, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5362310', 'n_tokens_mistral': 1507, 'n_tokens_neox': 1470, 'n_words': 292}