Merge pull request #9 from ASvyatkovskiy/from_cartesian

Section level support, n-gram features

dataformat/secformat_for_df.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+
+import simplejson
+import re
+
+#switch between primary key as a long integer and primary key as a string
+use_cryptic_pk = True
+
+finput = open("/scratch/network/alexeys/bills/lexs/bills_50000.json","r")
+
+#data first
+foutput = open("/scratch/network/alexeys/bills/lexs/sectioned_bills_50000.json","wa")
+
+#step through JSON. Tokenize "content" string
+sec_output_dicts = list()
+for line in finput.readlines():
+    output_dict = simplejson.loads(line)
+    sections = re.split("SECTION \d|section \d",output_dict['content']) 
+    filtered = filter(lambda x: len(x) > 170,sections)
+    for j,section in enumerate(filtered):
+        sec_output_dict = {'primary_key':'', 'content':None} 
+
+        sec_output_dict['content'] = section
+        sec_output_dict['primary_key'] = str(j)+"_"+output_dict['primary_key']
+        sec_output_dicts.append(sec_output_dict)
+
+for i, sec_output_dict in enumerate(sec_output_dicts):
+    if not use_cryptic_pk: sec_output_dict['primary_key'] = i
+    simplejson.dump(sec_output_dict, foutput)
+    foutput.write('\n')
+
+#metadata
+#FIXME yes...
+finput = open("/scratch/network/alexeys/bills/lexs/bills_50000.json","r")
+finput_meta = open("/scratch/network/alexeys/bills/lexs/bills_metadata_50000.json","r")
+foutput_meta = open("/scratch/network/alexeys/bills/lexs/sectioned_bills_metadata_50000.json","wa")
+
+#step through JSON. Tokenize "content" string
+sec_output_dicts_meta = list()
+for line_meta, line in zip(finput_meta.readlines(),finput.readlines()):
+    output_dict_meta = simplejson.loads(line_meta)
+    output_dict = simplejson.loads(line)
+    sections = re.split("SECTION \d|section \d",output_dict['content'])
+    filtered = filter(lambda x: len(x) > 170,sections)
+    for j in range(len(filtered)):
+        sec_output_dict_meta = {'year':None, 'state':'','docid':'', 'docversion':'', 'primary_key':''}
+        sec_output_dict_meta['primary_key'] = str(j)+"_"+output_dict_meta['primary_key']
+        sec_output_dict_meta['year'] = output_dict_meta['year']
+        sec_output_dict_meta['state'] = output_dict_meta['state']
+        sec_output_dict_meta['docid'] = output_dict_meta['docid']
+        sec_output_dict_meta['docversion'] = output_dict_meta['docversion']
+        sec_output_dicts_meta.append(sec_output_dict_meta)
+
+for i, sec_output_dict_meta in enumerate(sec_output_dicts_meta):
+    if not use_cryptic_pk: sec_output_dict_meta['primary_key'] = i
+    simplejson.dump(sec_output_dict_meta, foutput_meta)
+    foutput_meta.write('\n')

src/main/resources/adhocAnalyzer.conf

@@ -1,7 +1,9 @@
 adhocAnalyzer {
-  numTextFeatures = 5000, 
+  numTextFeatures = 16384,
+  addNGramFeatures = false,
+  nGramGranularity = 2,
   measureName = "cosine", 
-  inputBillsFile = "file:///scratch/network/alexeys/bills/lexs/bills_3b.json", 
-  inputPairsFile = "/user/alexeys/valid_pairs",
-  outputMainFile = "/user/alexeys/test_main_output"
+  inputBillsFile = "file:///scratch/network/alexeys/bills/lexs/sectioned_bills_50000.json",
+  inputPairsFile = "/user/alexeys/valid_section_pairs_50000test",
+  outputMainFile = "/user/alexeys/sectioned_output_cosine_50000test"
 }

src/main/resources/makeCartesian.conf

@@ -1,9 +1,10 @@
 makeCartesian {
   nPartitions  = 30,
-  docVersion   = "Enacted",
-  inputFile    = "file:///scratch/network/alexeys/bills/lexs/bills_metadata_3b.json",
-  outputFile   = "/user/alexeys/valid_pairs",
-  use_strict   = true,
+  docVersion   = "Introduced",
+  onlyInOut    = false, 
+  inputFile    = "file:///scratch/network/alexeys/bills/lexs/sectioned_bills_metadata_50000.json",
+  outputFile   = "/user/alexeys/valid_section_pairs_50000test",
+  use_strict   = false,
   strict_state = 5,
   strict_docid = "HB1001",
   strict_year  = 1991

src/main/scala/AdhocAnalyzer.scala

@@ -2,6 +2,8 @@
 
 Following parameters need to be filled in the resources/adhocAnalyzer.conf file:
     numTextFeatures: Number of text features to keep in hashingTF
+    addNGramFeatures: Boolean flag to indicate whether to add n-gram features
+    nGramGranularity: granularity of a rolling n-gram
     measureName: Similarity measure used
     inputBillsFile: Bill input file, one JSON per line
     inputPairsFile: CartesianPairs object input file
@@ -33,9 +35,6 @@ import org.apache.spark.mllib.linalg.{DenseVector, SparseVector, Vector, Vectors
 
 object AdhocAnalyzer {
 
-  /*
-    Experimental
-  */
   def converted(row: scala.collection.Seq[Any]) : Tuple2[String,SparseVector] = { 
     val ret = row.asInstanceOf[WrappedArray[Any]]
     val first = ret(0).asInstanceOf[String]
@@ -59,6 +58,10 @@ object AdhocAnalyzer {
     println("Elapsed time: " + (t1 - t0)/1000000000 + "s")
   }
 
+  def appendFeature(a: WrappedArray[String], b: WrappedArray[String]) : WrappedArray[String] = {
+     a ++ b
+  }  
+
   def run(params: Config) {
 
     val conf = new SparkConf().setAppName("AdhocAnalyzer")
@@ -85,24 +88,31 @@ object AdhocAnalyzer {
 
     //remove stopwords 
     var remover = new StopWordsRemover().setInputCol("words").setOutputCol("filtered")
-    val filtered_df = remover.transform(tokenized_df).drop("words")
+    var prefeaturized_df = remover.transform(tokenized_df).drop("words")
+
+    if (params.getBoolean("adhocAnalyzer.addNGramFeatures")) {
 
-    //ngram = NGram(n=2, inputCol="filtered", outputCol="ngram")
-    //ngram_df = ngram.transform(tokenized_df)
+       val ngram = new NGram().setN(params.getInt("adhocAnalyzer.nGramGranularity")).setInputCol("filtered").setOutputCol("ngram")
+       val ngram_df = ngram.transform(prefeaturized_df)
+
+       def appendFeature_udf = udf(appendFeature _)
+       prefeaturized_df = ngram_df.withColumn("combined", appendFeature_udf(col("filtered"),col("ngram"))).drop("filtered").drop("ngram").drop("cleaned")
+    } else {
+       prefeaturized_df = prefeaturized_df.select(col("primary_key"),col("filtered").alias("combined"))
+       prefeaturized_df.printSchema()
+    }
 
     //hashing
-    var hashingTF = new HashingTF().setInputCol("filtered").setOutputCol("rawFeatures").setNumFeatures(params.getInt("adhocAnalyzer.numTextFeatures"))
-    val featurized_df = hashingTF.transform(filtered_df).drop("filtered")
+    var hashingTF = new HashingTF().setInputCol("combined").setOutputCol("rawFeatures").setNumFeatures(params.getInt("adhocAnalyzer.numTextFeatures"))
+    val featurized_df = hashingTF.transform(prefeaturized_df)
 
     var idf = new IDF().setInputCol("rawFeatures").setOutputCol("pre_features")
     //val Array(train, cv) = featurized_df.randomSplit(Array(0.7, 0.3))
     var idfModel = idf.fit(featurized_df)
     val rescaled_df = idfModel.transform(featurized_df).drop("rawFeatures")
-    rescaled_df.show(5)
 
     val hashed_bills = featurized_df.select("primary_key","rawFeatures").rdd.map(row => converted(row.toSeq))
 
-    //Experimental
     //First, run the hashing step here
     val cartesian_pairs = spark.objectFile[CartesianPair](params.getString("adhocAnalyzer.inputPairsFile")).map(pp => (pp.pk1,pp.pk2))
 

src/main/scala/MakeCartesian.scala

@@ -4,6 +4,7 @@ Application: MakeCartesian, produce all the pairs of primary keys of the documen
 Following parameters need to be filled in the resources/makeCartesian.conf file:
 	docVersion: document version: consider document pairs having a specific version. E.g. Introduced, Enacted...
 	nPartitions: number of partitions in bills_meta RDD
+        onlyInOut: a switch between in-out of state and using both in-out and in-in state pairs
 	use_strict: boolean, yes or no to consider strict parameters
 	strict_state: specify state (a long integer from 1 to 50) for one-against-all user selection (strict)
 	strict_docid: specify document ID for one-against-all user selection (strict)
@@ -33,7 +34,7 @@ import scala.collection.mutable.WrappedArray
 
 object MakeCartesian {
 
-  def pairup (document: MetaDocument, thewholething: org.apache.spark.broadcast.Broadcast[Array[MetaDocument]], strict_params: Tuple4[Boolean, Int, java.lang.String, Int]) : (MetaDocument, Array[CartesianPair]) = {
+  def pairup (document: MetaDocument, thewholething: org.apache.spark.broadcast.Broadcast[Array[MetaDocument]], strict_params: Tuple4[Boolean, Int, java.lang.String, Int], onlyInOut: Boolean) : (MetaDocument, Array[CartesianPair]) = {
 
     val documents = thewholething.value
 
@@ -62,12 +63,20 @@ object MakeCartesian {
            }
          } 
        } else {
-         //simple condition
-          if (pk1 < pk2 && istate != jstate) {
-             var output: CartesianPair = CartesianPair(pk1,pk2)
-             output_arr += output
-          }
-       }
+          //simple condition
+          if (onlyInOut) {
+             if (pk1 < pk2 && istate != jstate) {
+                var output: CartesianPair = CartesianPair(pk1,pk2)
+                output_arr += output
+             }
+           } else {
+             //in-out and in-in
+             if (pk1 < pk2) {
+                var output: CartesianPair = CartesianPair(pk1,pk2)
+                output_arr += output
+             }
+           } 
+        }
      }
      (document,output_arr.toArray)
   }
@@ -106,7 +115,7 @@ object MakeCartesian {
 
     //will be array of tuples, but the keys are unique
     var cartesian_pairs = bills_meta.rdd.repartition(params.getInt("makeCartesian.nPartitions"))
-                          .map(x => pairup(x,bills_meta_bcast, strict_params))
+                          .map(x => pairup(x,bills_meta_bcast, strict_params, params.getBoolean("makeCartesian.onlyInOut")))
                           .filter({case (dd,ll) => (ll.length > 0)})
                           .map({case(k,v) => v}).flatMap(x => x) //.groupByKey()