<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Subject patterns with Spark
In this notebook we look at a large CSV file containing subject headings from MARC records.  Our goal is to process a file containing over eight million rows; because of this size, Spark might handle the volume more efficiently than R or Python, the other tools I would normally reach for when summarizing CSV data.
We begin by noting first that this is a PySpark notebook (if you're executing this in Jupyter, look for "pySpark" in the top right corner), not an ordinary Python notebook.  With a PySpark notebook, the pyspark command is executed in the background, giving you a SparkContext to work with (see Spark programming guide for details).
Let's verify that it all started correctly by checking that we have that SparkContext object available, which should present as sc
Step1:  Great, we're good to go.  Now let's take a look at the data.  See the README in this directory for background on how we generated the CSV file we'll work with here.
Step2:  Okay, that's a pretty big file
Step3:  It might be obvious from the header row, but we're looking at record identifiers, the two indicators, all of the subfields present in the subject represented as a single string (e.g. "ay0" means subfields a, y, and 0 are present with values), and the value of subfield 2.  You might notice that the bibid values repeat; in this case we are looking at three different subjects for bibid 2164503, four for 2164506, and two for 2164508.  Perfectly normal.
We can start to explore this data with Spark as-is, but let's create a smaller sample set just so we can get our logic right first.
Step4:  Looks good, but we lost our record ordering - not a big deal, but it might be helpful to get them back in order.  csvkit's csvsort can help with that
Step5:  We can confirm that our sample has at least some examples of multiple subjects per record, then, by noting that bibid 2091975 appears twice.  Good enough for now - the sorting won't matter to Spark, which will be counting in parallel anyway, but it's reassuring that our sample is useful.
Loading the CSV into an RDD and counting
Let's load our sample data into an RDD and pull up some basic counts.
Step6:  And verify that the data loaded correctly
Step7:  Looks good.  Let's split the data up now, and start to count things up.  First we have to parse the CSV structure, then we pull out the tag (index position 1) and put that into a tuple with the value 1.
Step8:  Now we have a long list of tuples like [('650', 1), ('651', 1), ('650', 1), ...].  We structure the data this way because that makes it easy to run this next step, adding it all up, using the Python operator "add" within the reduceByKey method.  We pipe that result into takeOrdered to get a sorted list of tags by count.  Note that we use -v for a descending count.
Step9:  Let's turn that into a function that we can use for the whole dataset.
Step10:  Now getting the count across all the data is just a matter of loading the full set in and passing it to the function.
Step11:  What exactly did Spark just do?
One of the things I like most about Spark is its web UI, which offers lots of insight about how jobs get processed. Even a counting process as simple as this yields interesting results.
Step12:  This shows us the several stages that went into the job in the timeline at top - the two skinny blue blobs at far right are the completed stages, one after the other.  You can see them spelled out in detail in the "Completed Stages" table at the bottom.  reduceByKey came first, and took 33 seconds, with six tasks.  You can see how the stages flowed with the data partitioning across cores in the "DAG Visualization" image in the middle.
Clicking on the timeline's blue blob representing the reduceByKey stage yields a more detail view of how that stage executed
Step13:  At the individual stage of reduceByKey, we can see the graph of partitioning the data across the four cores, then ultimately moving the data into the pair RDD (the tuples with the counts that got added up).  The timeline below shows those reduce jobs occuring across the cores, with the data partitioned into four jobs over approximately 1,400,000 records and then another two tasks across approximately 1,300,000 records, each of which starts (look close at the green bars) as soon as the first and second of the earlier tasks complete.  If this laptop had six cores or more, we can imagine that Spark would have executed them all simulaneously.
It is certainly possible that a decent RDBMS with appropriate indexes would have returned the same result in subsecond time, or far less than 33 seconds, in any case.  On the other hand, if we had 80,000,000 million records, or 8,000,000,000, and enough cores, Spark could blow away the database past a certain threshhold we'd cross in between there somewhere.
Counting tag/indicator/subfield co-occurences
We can repeat this process to count co-occurences of tags with indicators and subfields.  Let's see how many of each we have, but first, let's parameterize the field extractor piece so our function is more general.  While we're at it let's make the 25 item limit a parameter as well.
Step14:  It works - so now we can vary the columns it's using to count things up just by passing in a different splitter.  Let's look at just tag + indicator combinations first.
Step15:  It seems to be working just fine, so let's run it over the full dataset, pulling out the top combinations.
Step16:  Let's take a better look at that using matplotlib.
Step17:  Let's go a little further and add the subfields to the mix.
Step18:  This would look a lot better with the y-tick labels aligned to the left, but I couldn't quite get that right without some awkward code.  (Oh well... not the main point here.)
We can see from this plot that the tag+indicator+subfield combo that occurred the most, in more than 1 out of every 8 subjects, more than twice any other combo, was a 650 with blank first indicator and second indicator 7. According to the LC Concise MARC Bibliographic definition for 650 this indicates that the source of the term is specified in subfield 2.  Looking a few ticks further down, we see that the 2nd, 4th, 5th, 6th, 7th, 8th, and 9th highest patterns also included indicator 7, which holds the same meaning for topic and geographic (651) added entries, as well as genre/form (655) and chronological added entries (648). 
So let's carry on and count the subfield 2 values.
Step19:  I'll leave scraping LC's Subject Heading and Source Term Codes and mapping the dl element values to a table we can cross-reference as an exercise for the reader. 
Step20:  A majority of subjects use indicator 2 of '7', that a source is specified in subfield 2.  Let's look at that next.  
In second place is LCSH ('0'), followed by Répertoire de vedettes-matière ('6'), not specified ('4'), MeSH ('2'), LCSH for children's lit ('1'), '8' (what's that?), CSH ('5'), NAL ('3'), and '9' (another what's that?).  The 'i2' value is from the header line.
So, about those subfield 2 values; let's specifically filter for indicator 2 == '7' and try again.  The result should look similar to what we saw a few cells back, but this should be more precise, in theory.
Step21:  Indeed, it is almost exactly the same, with just a few counts off by a few items.
Descriptive statistics
<ASSISTANT_TASK:>
Python Code:
sc
Explanation: Subject patterns with Spark
In this notebook we look at a large CSV file containing subject headings from MARC records.  Our goal is to process a file containing over eight million rows; because of this size, Spark might handle the volume more efficiently than R or Python, the other tools I would normally reach for when summarizing CSV data.
We begin by noting first that this is a PySpark notebook (if you're executing this in Jupyter, look for "pySpark" in the top right corner), not an ordinary Python notebook.  With a PySpark notebook, the pyspark command is executed in the background, giving you a SparkContext to work with (see Spark programming guide for details).
Let's verify that it all started correctly by checking that we have that SparkContext object available, which should present as sc:
End of explanation
!ls -lh combined.csv
!wc combined.csv
Explanation: Great, we're good to go.  Now let's take a look at the data.  See the README in this directory for background on how we generated the CSV file we'll work with here.
End of explanation
!head combined.csv
Explanation: Okay, that's a pretty big file:  191M of uncompressed CSV, representing 8,357,060 individual subjects across all the MARC records we started from.  In case you're wondering, here's what a few subjects look like:
End of explanation
!gshuf -n 100000 -o sample.csv combined.csv
!wc sample.csv
!head sample.csv
Explanation: It might be obvious from the header row, but we're looking at record identifiers, the two indicators, all of the subfields present in the subject represented as a single string (e.g. "ay0" means subfields a, y, and 0 are present with values), and the value of subfield 2.  You might notice that the bibid values repeat; in this case we are looking at three different subjects for bibid 2164503, four for 2164506, and two for 2164508.  Perfectly normal.
We can start to explore this data with Spark as-is, but let's create a smaller sample set just so we can get our logic right first.
End of explanation
!csvsort --no-header-row -c1 sample.csv > sample-sorted.csv
!head sample-sorted.csv
Explanation: Looks good, but we lost our record ordering - not a big deal, but it might be helpful to get them back in order.  csvkit's csvsort can help with that:
End of explanation
subjects_sample = sc.textFile("sample.csv")
Explanation: We can confirm that our sample has at least some examples of multiple subjects per record, then, by noting that bibid 2091975 appears twice.  Good enough for now - the sorting won't matter to Spark, which will be counting in parallel anyway, but it's reassuring that our sample is useful.
Loading the CSV into an RDD and counting
Let's load our sample data into an RDD and pull up some basic counts.
End of explanation
subjects_sample.count()
Explanation: And verify that the data loaded correctly:
End of explanation
tag_pairs = subjects_sample.map(lambda l: l.split(",")).map(lambda l: (l[1], 1))
Explanation: Looks good.  Let's split the data up now, and start to count things up.  First we have to parse the CSV structure, then we pull out the tag (index position 1) and put that into a tuple with the value 1.
End of explanation
from operator import add
tag_counts = tag_pairs.reduceByKey(add).takeOrdered(25, key=lambda (k, v): -v)
tag_counts
Explanation: Now we have a long list of tuples like [('650', 1), ('651', 1), ('650', 1), ...].  We structure the data this way because that makes it easy to run this next step, adding it all up, using the Python operator "add" within the reduceByKey method.  We pipe that result into takeOrdered to get a sorted list of tags by count.  Note that we use -v for a descending count.
End of explanation
def count_tags(subjects_rdd):
    pairs = subjects_rdd.map(lambda l: l.split(',')).map(lambda l: (l[1], 1))
    return pairs.reduceByKey(add).takeOrdered(25, key=lambda (k, v): -v)                                                        
Explanation: Let's turn that into a function that we can use for the whole dataset.
End of explanation
subjects_all = sc.textFile("combined.csv")
%time count_tags(subjects_all)
Explanation: Now getting the count across all the data is just a matter of loading the full set in and passing it to the function.
End of explanation
from IPython.display import Image
Image("ui-job.png")
Explanation: What exactly did Spark just do?
One of the things I like most about Spark is its web UI, which offers lots of insight about how jobs get processed. Even a counting process as simple as this yields interesting results.
End of explanation
Image("ui-stages.png")
Explanation: This shows us the several stages that went into the job in the timeline at top - the two skinny blue blobs at far right are the completed stages, one after the other.  You can see them spelled out in detail in the "Completed Stages" table at the bottom.  reduceByKey came first, and took 33 seconds, with six tasks.  You can see how the stages flowed with the data partitioning across cores in the "DAG Visualization" image in the middle.
Clicking on the timeline's blue blob representing the reduceByKey stage yields a more detail view of how that stage executed:
End of explanation
def counter(subjects_rdd, splitter, limit=25):
    pairs = subjects_rdd.map(lambda l: l.split(',')).map(splitter)
    return pairs.reduceByKey(add).takeOrdered(limit, key=lambda (k, v): -v)  
splitter = lambda l: (l[1], 1)
counter(subjects_sample, splitter, 10)
Explanation: At the individual stage of reduceByKey, we can see the graph of partitioning the data across the four cores, then ultimately moving the data into the pair RDD (the tuples with the counts that got added up).  The timeline below shows those reduce jobs occuring across the cores, with the data partitioned into four jobs over approximately 1,400,000 records and then another two tasks across approximately 1,300,000 records, each of which starts (look close at the green bars) as soon as the first and second of the earlier tasks complete.  If this laptop had six cores or more, we can imagine that Spark would have executed them all simulaneously.
It is certainly possible that a decent RDBMS with appropriate indexes would have returned the same result in subsecond time, or far less than 33 seconds, in any case.  On the other hand, if we had 80,000,000 million records, or 8,000,000,000, and enough cores, Spark could blow away the database past a certain threshhold we'd cross in between there somewhere.
Counting tag/indicator/subfield co-occurences
We can repeat this process to count co-occurences of tags with indicators and subfields.  Let's see how many of each we have, but first, let's parameterize the field extractor piece so our function is more general.  While we're at it let's make the 25 item limit a parameter as well.
End of explanation
counter(subjects_sample, lambda l: ((l[1], l[2], l[3]), 1), 5)
Explanation: It works - so now we can vary the columns it's using to count things up just by passing in a different splitter.  Let's look at just tag + indicator combinations first.
End of explanation
top_subjects_indicators = counter(subjects_all, lambda l: ((l[1], l[2], l[3]), 1))
Explanation: It seems to be working just fine, so let's run it over the full dataset, pulling out the top combinations.
End of explanation
%matplotlib inline
top = [("%s-%s-%s" % (k[0], '_' if k[1] == ' ' else k[1], '_' if k[2] == ' ' else k[2]), v) 
         for k, v in top_subjects_indicators]
import matplotlib.pyplot as plt
import numpy as np
y_pos = np.arange(len(top))
plt.barh(y_pos, [v for k, v in reversed(top)], 
         align='center', color='steelblue', edgecolor='none')
plt.xlabel('Count')
plt.xlim([0, top[0][1] * 1.05])
plt.gca().xaxis.grid(True)
plt.ylabel('Tag and indicators')
plt.yticks(y_pos, [k for k, v in reversed(top)])
plt.ylim([-1, 25])
plt.tick_params(axis='y', labelsize=14)
plt.title('Counts of tags + indicator combinations')
plt.show()
Explanation: Let's take a better look at that using matplotlib.
End of explanation
top_subfields = counter(subjects_all, lambda l: ((l[1], l[2], l[3], l[4]), 1))
top = [("%s-%s-%s-%s" % (k[0], '_' if k[1] == ' ' else k[1], '_' if k[2] == ' ' else k[2], k[3]), v) 
       for k, v in top_subfields]
y_pos = np.arange(len(top_subf))
plt.barh(y_pos, [v for k, v in reversed(top)], 
         align='center', color='steelblue', edgecolor='none')
plt.xlabel('Count')
plt.xlim([0, top[0][1] * 1.05])
plt.gca().xaxis.grid(True)
plt.ylabel('Tag and indicators')
plt.yticks(y_pos, [k for k, v in reversed(top)])
plt.ylim([-1, 25])
plt.tick_params(axis='y', labelsize=14)
plt.title('Counts of tags + indicator combinations')
plt.show()
Explanation: Let's go a little further and add the subfields to the mix.
End of explanation
sf2 = counter(subjects_all, lambda l: (l[5], 1))
sf2
Explanation: This would look a lot better with the y-tick labels aligned to the left, but I couldn't quite get that right without some awkward code.  (Oh well... not the main point here.)
We can see from this plot that the tag+indicator+subfield combo that occurred the most, in more than 1 out of every 8 subjects, more than twice any other combo, was a 650 with blank first indicator and second indicator 7. According to the LC Concise MARC Bibliographic definition for 650 this indicates that the source of the term is specified in subfield 2.  Looking a few ticks further down, we see that the 2nd, 4th, 5th, 6th, 7th, 8th, and 9th highest patterns also included indicator 7, which holds the same meaning for topic and geographic (651) added entries, as well as genre/form (655) and chronological added entries (648). 
So let's carry on and count the subfield 2 values.
End of explanation
pairs = subjects_all.map(lambda l: l.split(','))
pairs = pairs.filter(lambda l: l[3] != ' ')
pairs = pairs.map(lambda l: (l[3], 1))
pairs.reduceByKey(add).takeOrdered(25, key=lambda (k, v): -v)
Explanation: I'll leave scraping LC's Subject Heading and Source Term Codes and mapping the dl element values to a table we can cross-reference as an exercise for the reader. :)
But we can see that nearly half of the subjects did not reference a source, more than a quarter referenced OCLC's FAST, and quite a few came from German (swd, gnd), Dutch (gtt), and French (ram).
Let's take a closer look at the use of indicator 2.  What's present?
End of explanation
sf2 = subjects_all.map(lambda l: l.split(','))
sf2 = sf2.filter(lambda l: l[3] == '7')
sf2 = sf2.map(lambda l: (l[5], 1))
sf2.reduceByKey(add).takeOrdered(100, key=lambda (k, v): -v)
Explanation: A majority of subjects use indicator 2 of '7', that a source is specified in subfield 2.  Let's look at that next.  
In second place is LCSH ('0'), followed by Répertoire de vedettes-matière ('6'), not specified ('4'), MeSH ('2'), LCSH for children's lit ('1'), '8' (what's that?), CSH ('5'), NAL ('3'), and '9' (another what's that?).  The 'i2' value is from the header line.
So, about those subfield 2 values; let's specifically filter for indicator 2 == '7' and try again.  The result should look similar to what we saw a few cells back, but this should be more precise, in theory.
End of explanation
from pyspark.mllib.stat import Statistics
Explanation: Indeed, it is almost exactly the same, with just a few counts off by a few items.
Descriptive statistics
End of explanation
<END_TASK>