<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Example 1
Step1:  This dataset is a debug dump from a Lustre filesystem.  Typically these events occur due to code bugs (LBUG), heavy load, hardware problems, or misbehaving user application IO.
Let's analyze some of the log structure to determine what may have caused this debug dump.
Step2:  Let's take a look at the first five lines of the debug log.  This log is colon-delimited, and roughly corresponds to the following information
Step3:  Now let's split each line of the RDD into lowercase "words".
Lambda functions are ubiquitous in Spark, I presume due to the functional programming underpinnings of Scala.  They act on each partition in parallel, and operate on each line.
Step4:  Notice that this map returns immediately; no actions have been taken- the DAG has been updated to prepare for transformations.  I like to think of this as analogous to a page fault, but applying to a Directed Acyclic Graph.
Step5:  Part 2
Step6:  Did it work?
Step7:  Now we issue an action to the RDD
Step8:  And as a percent of the overall file?
Step9:  Now let's determine the effect of the flatMap
Step10:  Now filter out "words" longer than 2 characters.
Step11:  To sort words by number of occurences we map each word of each line to a tuple
Step12:  We utilize reduceByKey
Step13:  We swap the order of the tuple's contents to sort by the number rather than words.  The argument "False" passed to sortByKey instructs it to sort descending.
<ASSISTANT_TASK:>
Python Code:
from pyspark import SparkConf, SparkContext
import re
Explanation: Example 1: Parallel Log Parsing with Map and Filter
Step 1: Data ingest and parsing
End of explanation
sc
partitions = 18
parlog = sc.textFile("/lustre/janus_scratch/dami9546/lustre_debug.out", partitions)
Explanation: This dataset is a debug dump from a Lustre filesystem.  Typically these events occur due to code bugs (LBUG), heavy load, hardware problems, or misbehaving user application IO.
Let's analyze some of the log structure to determine what may have caused this debug dump.
End of explanation
parlog.take(5)
Explanation: Let's take a look at the first five lines of the debug log.  This log is colon-delimited, and roughly corresponds to the following information: 
0-1 describe subsystem ID
2
3 timestamp
4-6 PIDs
7 relevant code module
8 code line
9 function and message
End of explanation
words = parlog.map(lambda line: re.split('\W+', line.lower().strip()))
Explanation: Now let's split each line of the RDD into lowercase "words".
Lambda functions are ubiquitous in Spark, I presume due to the functional programming underpinnings of Scala.  They act on each partition in parallel, and operate on each line.
End of explanation
words.take(2)
Explanation: Notice that this map returns immediately; no actions have been taken- the DAG has been updated to prepare for transformations.  I like to think of this as analogous to a page fault, but applying to a Directed Acyclic Graph.
End of explanation
mfds = words.filter(lambda x: 'mfd' and 'change' in x)
Explanation: Part 2: Counting Occurences
My experience with Lustre affords me (some) insight into this- I know the system has been susceptible to MDS overloading due to applications creating tons of small files, or issuing lots of MDS RPCs.  I want to look for all lines that contain mfd changes.
Let's apply a filter to this RDD.  Let's create a new RDD that only contains lines with mfs changes.
End of explanation
mfds.take(2)
Explanation: Did it work?
End of explanation
mfds.count()
Explanation: Now we issue an action to the RDD: the DAG performs the lazily executed functions.  In this case we count the number of lines in the mfds RDD.
End of explanation
'{0:0.2f}%'.format((mfds.count()/float(parlog.count()))*100)
Explanation: And as a percent of the overall file?
End of explanation
flatwords = parlog.flatMap(lambda line: re.split('\W+', line.lower().strip()))
Explanation: Now let's determine the effect of the flatMap: this behaves like map, but does not return a list for each line.  Rather, it aggregates (flattens) the output into a single list.
End of explanation
longwords = flatwords.filter(lambda x: len(x) > 2 )
longwords.take(10)
Explanation: Now filter out "words" longer than 2 characters.
End of explanation
longwords = longwords.map(lambda word: (word, 1))
Explanation: To sort words by number of occurences we map each word of each line to a tuple: itself and 1.  We will perform a reduction on these tuples to get counts.
End of explanation
longcount = longwords.reduceByKey(lambda a, b: a + b)
longcount.take(10)
Explanation: We utilize reduceByKey: this operation performs a function on identical keys.  By default this will be the first element of the tuple.  Since this will be the word, the behavior is desired.
Note that reduce operations are accumulators and must be associative.
End of explanation
longwords = longcount.map(lambda x: (x[1], x[0])).sortByKey(False)
longwords.take(20)
Explanation: We swap the order of the tuple's contents to sort by the number rather than words.  The argument "False" passed to sortByKey instructs it to sort descending.
End of explanation
<END_TASK>