Apache Spark – A Deep Dive – Series 4 of N – Using FlatMap RDDS, Regular Expressions and Sorting of RDDS

• E.g. Lets imagine that a file ‘test.txt’ has the below 4 lines:
  • I am Mohammad Naeem.
    I am 39 years old.
    I live in Texas, USA.
    I am software engineer.
  • If we read this file using Map ( having 4 lines)as:
    • l= sc.textFile(“file:///D:/dumps/BigData/Otherdata/test.txt”)
      lRDD =l.map(lambda line: line.split())
    • the new RDD lRDD will also have 4 rows just like ‘l’ the intial RDD
    • Each row will have a tuple of splited fields
  • If we read this file using FlatMap ( having 4 lines)as:
    • l= sc.textFile(“file:///D:/dumps/BigData/Otherdata/test.txt”)
      lRDD =l.flatMap(lambda line: line.split())
    • the new RDD lRDD will have more than 4 lines(may be each each splited word is a new line)
    • Each row will have a tuple of splited fields
  • So FlatMap function is a
    • one to many relationship
    • thus performs faster than Map

• from pyspark import SparkConf, SparkContext sConf = SparkConf().setMaster(“local”).setAppName(“FlatMapRDDApp”) sContext = SparkContext(conf = sConf) fileData = sContext.textFile(“file:///D:/dumps/BigData/Otherdata/LinuxRefresher.txt”) # flatMap will split the fileData into a one to many relationship fileFlatMapRDD =fileData.flatMap(lambda line: line.split()) # countByValue will create a tuple like (word, count) resultData= fileFlatMapRDD.countByValue() for result in resultData.items(): print(“%s: %i” % (result[0], result[1]))

• Here is the output:
• 
• We can see that for each word it is showing the count of the word. But there are lot of issues in the data. ‘!’, ‘~’ or apostrophe(‘) are all counted as words. If we can use a regular expression to clean up the data, the output will be much better.

from pyspark import SparkConf, SparkContext # import regular expression module import re # write a function to clean up words def cleanupWords(word): return re.compile(r’\W+’, re.UNICODE).split(word.lower()) sConf = SparkConf().setMaster(“local”).setAppName(“FlatMapWithRegularExprRDDApp”) sContext = SparkContext(conf = sConf) fileData = sContext.textFile(“file:///D:/dumps/BigData/Otherdata/LinuxRefresher.txt”) # flatMap will split the fileData into a one to many relationship fileFlatMapRDD =fileData.flatMap(cleanupWords) # countByValue will create a tuple like (word, count) resultData= fileFlatMapRDD.countByValue() for result in resultData.items(): print(“%s: %i” % (result[0], result[1]))

from pyspark import SparkConf, SparkContext # import regular expression module import re # write a function to clean up words def cleanupWords(word): return re.compile(r’\W+’, re.UNICODE).split(word.lower()) sConf = SparkConf().setMaster(“local”).setAppName(“FlatMapWithRegularExprRDDApp”) sContext = SparkContext(conf = sConf) fileData = sContext.textFile(“file:///D:/dumps/BigData/Otherdata/LinuxRefresher.txt”) # flatMap will split the fileData into a one to many relationship fileFlatMapRDD =fileData.flatMap(cleanupWords) # Now that we have got the words flat mapped, we have to do the following # Step 1 – create another RDD with each line as a tuple but in the form (word, 1) # Step 2 – now we can aggregate by adding the counts wordsWithCountsRDD = fileFlatMapRDD.map(lambda x: (x,1)).reduceByKey(lambda x, y: x + y) # Step 3 – now we can map again but we will flip the sequence # but this time the key would not be the word but the count e.g the tuple will be (x[1], x[0]) wordsWithCountsSortedRDD = wordsWithCountsRDD.map(lambda x: (x[1], x[0])).sortByKey(False) # collect the result to affect the RDD results = wordsWithCountsSortedRDD.collect() for result in results: count = result[0] word = result[1] print(“%s: %i” % (word, count))