from __future__ import print_function
import sys
from pyspark import SparkContext
from operator import add
from lib.csvline import Csvline

def get_headers(fn):
    csvline = Csvline()
    with open(fn, "r") as f:
        return csvline.parse(f.readline())


if __name__ == "__main__":

    fields = ["dwc:scientificName", "dwc:verbatimLocality"]
    #fields = ["dwc:scientificName"]

    fn = "data/0b17c21a-f7e2-4967-bdf8-60cf9b06c721/occurrence.txt"
    headers = get_headers(fn)

    sc = SparkContext(appName="TestUniqueCSVline")
    csvline = Csvline()

    records = sc.textFile(fn)
    #parsed = records.map(lambda x: csvline.parse(x.encode("utf8"), headers) )
    #parsed.cache()

    for field in fields:
        counts = records.map(lambda x: (csvline.parse(x.encode("utf8"), headers)[field], 1))
        #counts = parsed.map(lambda x: (x[field], 1))
        #sorted = counts.sortByKey()
        totals = counts.reduceByKey(add)
        totals.saveAsTextFile("out_{0}".format(field.replace(":", "_")))

#    output = totals.collect()
#    with open("unique-csvline.out", "w") as f:
#        for (word, count) in output:
#            f.write("{0},{1}\n".format(count, word.encode("utf8")))

@@ -42,16 +42,34 @@
 ort sys%0A
+import unicodecsv%0A
 from pys
@@ -337,70 +337,393 @@
 me%22,
- %22dwc:verbatimLocality%22%5D%0A    #fields = %5B%22dwc:scientificName%22%5D%0A
+%0A              %22dwc:specificEpithet%22,%0A              %22dwc:collectionCode%22,%0A              %22dwc:occurrenceID%22,%0A              %22dwc:institutionCode%22,%0A              %22dwc:genus%22,%0A              %22dwc:verbatimLocality%22,%0A              %22dwc:catalogNumber%22,%0A              %22dwc:eventDate%22,%0A              %22dwc:recordedBy%22%5D%0A%0A%0A    #fields = %5B%22dwc:occurrenceID%22%5D%0A%0A    #fn = %22data/idigbio/occurrence.csv%22
 %0A   
@@ -1278,24 +1278,25 @@
 dd)%0A        
+#
 totals.saveA
@@ -1349,17 +1349,20 @@
 %22_%22)))%0A%0A
-#
+    
     outp
@@ -1387,47 +1387,107 @@
 t()%0A
-#
     
-with open(%22unique-csvline.out%22
+    out_fn = %22out/unique_%7B0%7D.csv%22.format(field.replace(%22:%22, %22_%22))%0A        with open(out_fn
 , %22w
+b
 %22) a
@@ -1491,17 +1491,74 @@
 ) as f:%0A
-#
+            csvwriter = unicodecsv.writer(f, %22excel%22)%0A    
         
@@ -1586,17 +1586,20 @@
 output:%0A
-#
+    
         
@@ -1606,60 +1606,38 @@
     
-f.
+csv
 write
-(%22%7B0%7D,%7B1%7D%5Cn%22.format(count, word.encode(%22utf8%22))
+r.writerow(%5Bword, count%5D
 )%0A

Change to writing csv files for output, requires setting spark.driver.maxResultSize large enough for the collect() call to not run out of memory. Also using a larger list fo fields.