Added sys path to guarantee imports | Added SparkConf to all files

Pedro Bernardo · Pedro Bernardo · commit 942672bc09be · 2018-01-31T06:55:49.000-08:00
diff --git a/advanced/accumulator/StackOverFlowSurvey.py b/advanced/accumulator/StackOverFlowSurvey.py
@@ -1,25 +1,24 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
-def filterResponseFromCanada(response, total, missingSalaryMidPoint):
-    splits = Utils.COMMA_DELIMITER.split(response)
-    total.add(1)
-    if not splits[14]:
-        missingSalaryMidPoint.add(1)
-    return splits[2] == "Canada"
-
 if __name__ == "__main__":
-    sc = SparkContext("local", "StackOverFlowSurvey")
-    sc.setLogLevel("ERROR")
-
+    conf = SparkConf().setAppName('StackOverFlowSurvey').setMaster("local[*]")
+    sc = SparkContext(conf = conf)
     total = sc.accumulator(0)
     missingSalaryMidPoint = sc.accumulator(0)
-
     responseRDD = sc.textFile("in/2016-stack-overflow-survey-responses.csv")
 
-    responseFromCanada = responseRDD.filter(lambda response: \
-        filterResponseFromCanada(response, total, missingSalaryMidPoint))
+    def filterResponseFromCanada(response):
+        splits = Utils.COMMA_DELIMITER.split(response)
+        total.add(1)
+        if not splits[14]:
+            missingSalaryMidPoint.add(1)
+        return splits[2] == "Canada"
 
+    responseFromCanada = responseRDD.filter(filterResponseFromCanada)
     print("Count of responses from Canada: {}".format(responseFromCanada.count()))
     print("Total count of responses: {}".format(total.value))
-    print("Count of responses missing salary middle point: {}".format(missingSalaryMidPoint.value))
+    print("Count of responses missing salary middle point: {}" \
+        .format(missingSalaryMidPoint.value))
diff --git a/advanced/accumulator/StackOverFlowSurveyFollowUp.py b/advanced/accumulator/StackOverFlowSurveyFollowUp.py
@@ -1,26 +1,25 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
-def filterResponseFromCanada(response, total, missingSalaryMidPoint, processedBytes):
-    processedBytes.add(len(response.encode('utf-8')))
-    splits = Utils.COMMA_DELIMITER.split(response)
-    total.add(1)
-    if not splits[14]:
-        missingSalaryMidPoint.add(1)
-    return splits[2] == "Canada"
-
 if __name__ == "__main__":
-    sc = SparkContext("local", "StackOverFlowSurvey")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName('StackOverFlowSurvey').setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     total = sc.accumulator(0)
     missingSalaryMidPoint = sc.accumulator(0)
     processedBytes = sc.accumulator(0)
-
     responseRDD = sc.textFile("in/2016-stack-overflow-survey-responses.csv")
 
-    responseFromCanada = responseRDD.filter(lambda response: \
-        filterResponseFromCanada(response, total, missingSalaryMidPoint, processedBytes))
+    def filterResponseFromCanada(response):
+        processedBytes.add(len(response.encode('utf-8')))
+        splits = Utils.COMMA_DELIMITER.split(response)
+        total.add(1)
+        if not splits[14]:
+            missingSalaryMidPoint.add(1)
+        return splits[2] == "Canada"
+    responseFromCanada = responseRDD.filter(filterResponseFromCanada)
 
     print("Count of responses from Canada: {}".format(responseFromCanada.count()))
     print("Number of bytes processed: {}".format(processedBytes.value))
diff --git a/advanced/broadcast/UkMakerSpaces.py b/advanced/broadcast/UkMakerSpaces.py
@@ -1,19 +1,21 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
-def getPostPrefix(line: str):
-    splits = Utils.COMMA_DELIMITER.split(line)
-    postcode = splits[4]
-    return None if not postcode else postcode.split(" ")[0]
-
 def loadPostCodeMap():
     lines = open("in/uk-postcode.csv", "r").read().split("\n")
     splitsForLines = [Utils.COMMA_DELIMITER.split(line) for line in lines if line != ""]
     return {splits[0]: splits[7] for splits in splitsForLines}
 
+def getPostPrefix(line: str):
+    splits = Utils.COMMA_DELIMITER.split(line)
+    postcode = splits[4]
+    return None if not postcode else postcode.split(" ")[0]
+
 if __name__ == "__main__":
-    sc = SparkContext("local", "UkMakerSpaces")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName('UkMakerSpaces').setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     postCodeMap = sc.broadcast(loadPostCodeMap())
 
diff --git a/advanced/broadcast/UkMakerSpacesWithoutBroadcast.py b/advanced/broadcast/UkMakerSpacesWithoutBroadcast.py
@@ -1,26 +1,28 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
-def getPostPrefixes(line: str):
-    postcode = Utils.COMMA_DELIMITER.split(line)[4]
-    cleanedPostCode = postcode.replace("\\s+", "")
-    return [cleanedPostCode[0:i] for i in range(0,len(cleanedPostCode)+1)]
-
 def loadPostCodeMap():
     lines = open("in/uk-postcode.csv", "r").read().split("\n")
     splitsForLines = [Utils.COMMA_DELIMITER.split(line) for line in lines if line != ""]
     return {splits[0]: splits[7] for splits in splitsForLines}
 
+def getPostPrefix(line: str):
+    splits = Utils.COMMA_DELIMITER.split(line)
+    postcode = splits[4]
+    return None if not postcode else postcode.split(" ")[0]
+
 if __name__ == "__main__":
-    sc = SparkContext("local", "UkMakerSpaces")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName('UkMakerSpaces').setMaster("local[*]")
+    sc = SparkContext(conf = conf)
     postCodeMap = loadPostCodeMap()
     makerSpaceRdd = sc.textFile("in/uk-makerspaces-identifiable-data.csv")
 
     regions = makerSpaceRdd \
       .filter(lambda line: Utils.COMMA_DELIMITER.split(line)[0] != "Timestamp") \
-      .map(lambda line: next((postCodeMap[prefix] for prefix in getPostPrefixes(line) \
-      if prefix in postCodeMap), "Unknow"))
+      .map(lambda line: postCodeMap[getPostPrefix(line)] \
+        if getPostPrefix(line) in postCodeMap else "Unknow")
 
     for region, count in regions.countByValue().items():
         print("{} : {}".format(region, count))
diff --git a/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.py b/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.py
@@ -1,9 +1,8 @@
-from pyspark import SparkContext
+from pyspark import SparkContext, SparkConf
 
 if __name__ == "__main__":
-
-    sc = SparkContext("local", "AverageHousePrice")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName("AverageHousePrice").setMaster("local")
+    sc = SparkContext(conf = conf)
 
     lines = sc.textFile("in/RealEstate.csv")
     cleanedLines = lines.filter(lambda line: "Bedrooms" not in line)
diff --git a/pairRdd/aggregation/reducebykey/housePrice/AverageHousePriceSolution.py b/pairRdd/aggregation/reducebykey/housePrice/AverageHousePriceSolution.py
@@ -1,24 +1,26 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
+from pairRdd.aggregation.reducebykey.housePrice.AvgCount import AvgCount
 
 if __name__ == "__main__":
-
-    sc = SparkContext("local", "avgHousePrice")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName("avgHousePrice").setMaster("local[3]")
+    sc = SparkContext(conf = conf)
 
     lines = sc.textFile("in/RealEstate.csv")
     cleanedLines = lines.filter(lambda line: "Bedrooms" not in line)
 
     housePricePairRdd = cleanedLines.map(lambda line: \
-        (line.split(",")[3], (1, float(line.split(",")[2]))))
+        (line.split(",")[3], AvgCount(1, float(line.split(",")[2]))))
 
     housePriceTotal = housePricePairRdd \
-        .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))
+        .reduceByKey(lambda x, y: AvgCount(x.count + y.count, x.total + y.total))
 
     print("housePriceTotal: ")
-    for bedroom, total in housePriceTotal.collect():
-        print("{} : {}".format(bedroom, total))
+    for bedroom, avgCount in housePriceTotal.collect():
+        print("{} : ({}, {})".format(bedroom, avgCount.count, avgCount.total))
 
-    housePriceAvg = housePriceTotal.mapValues(lambda avgCount: avgCount[1] / avgCount[0])
+    housePriceAvg = housePriceTotal.mapValues(lambda avgCount: avgCount.total / avgCount.count)
     print("\nhousePriceAvg: ")
     for bedroom, avg in housePriceAvg.collect():
         print("{} : {}".format(bedroom, avg))
diff --git a/pairRdd/filter/AirportsNotInUsaSolution.py b/pairRdd/filter/AirportsNotInUsaSolution.py
@@ -1,10 +1,12 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
 if __name__ == "__main__":
 
-    sc = SparkContext("local", "airports")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName("airports").setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     airportsRDD = sc.textFile("in/airports.text")
 
diff --git a/pairRdd/groupbykey/AirportsByCountrySolution.py b/pairRdd/groupbykey/AirportsByCountrySolution.py
@@ -1,10 +1,12 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
 if __name__ == "__main__":
 
-    sc = SparkContext("local", "airports")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName("airports").setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     lines = sc.textFile("in/airports.text")
 
@@ -15,4 +17,4 @@
     airportsByCountry = countryAndAirportNameAndPair.groupByKey()
 
     for country, airportName in airportsByCountry.collectAsMap().items():
-        print("{}: {}".format(country,list(airportName)))
+        print("{}: {}".format(country, list(airportName)))
diff --git a/pairRdd/mapValues/AirportsUppercaseSolution.py b/pairRdd/mapValues/AirportsUppercaseSolution.py
@@ -1,10 +1,11 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
 if __name__ == "__main__":
-
-    sc = SparkContext("local", "airports")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName("airports").setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     airportsRDD = sc.textFile("in/airports.text")
 
diff --git a/pairRdd/sort/AverageHousePriceSolution.py b/pairRdd/sort/AverageHousePriceSolution.py
@@ -1,11 +1,11 @@
+import sys
+sys.path.insert(0, '.')
 from pairRdd.aggregation.reducebykey.housePrice.AvgCount import AvgCount
-from pyspark import SparkContext
-
+from pyspark import SparkContext, SparkConf
 
 if __name__ == "__main__":
-
-    sc = SparkContext("local", "averageHousePriceSolution")
-    sc.setLogLevel("ERROR")
+    conf = SparkConf().setAppName("averageHousePriceSolution").setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     lines = sc.textFile("in/RealEstate.csv")
     cleanedLines = lines.filter(lambda line: "Bedrooms" not in line)
diff --git a/rdd/airports/AirportsByLatitudeSolution.py b/rdd/airports/AirportsByLatitudeSolution.py
@@ -1,12 +1,15 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
 def splitComma(line: str):
     splits = Utils.COMMA_DELIMITER.split(line)
     return "{}, {}".format(splits[1], splits[6])
 
 if __name__ == "__main__":
-    sc = SparkContext("local", "airports")
+    conf = SparkConf().setAppName("airports").setMaster("local[*]")
+    sc = SparkContext(conf = conf)
     
     airports = sc.textFile("in/airports.text")
 
diff --git a/rdd/airports/AirportsInUsaSolution.py b/rdd/airports/AirportsInUsaSolution.py
@@ -1,12 +1,15 @@
-from pyspark import SparkContext
+import sys
+sys.path.insert(0, '.')
+from pyspark import SparkContext, SparkConf
 from commons.Utils import Utils
 
 def splitComma(line: str):
     splits = Utils.COMMA_DELIMITER.split(line)
     return "{}, {}".format(splits[1], splits[2])
 
 if __name__ == "__main__":
-    sc = SparkContext("local", "airports")
+    conf = SparkConf().setAppName("airports").setMaster("local[*]")
+    sc = SparkContext(conf = conf)
 
     airports = sc.textFile("in/airports.text")
     airportsInUSA = airports.filter(lambda line : Utils.COMMA_DELIMITER.split(line)[3] == "\"United States\"")
diff --git a/rdd/count/CountExample.py b/rdd/count/CountExample.py
@@ -3,9 +3,12 @@
 if __name__ == "__main__":
     conf = SparkConf().setAppName("count").setMaster("local[*]")
     sc = SparkContext(conf = conf)
+    
     inputWords = ["spark", "hadoop", "spark", "hive", "pig", "cassandra", "hadoop"]
+    
     wordRdd = sc.parallelize(inputWords)
     print("Count: {}".format(wordRdd.count()))
+    
     worldCountByValue = wordRdd.countByValue()
     print("CountByValue: ")
     for word, count in worldCountByValue.items():
diff --git a/sparkSql/HousePriceProblem.py b/sparkSql/HousePriceProblem.py
@@ -4,20 +4,21 @@
     Create a Spark program to read the house data from in/RealEstate.csv,
     group by location, aggregate the average price per SQ Ft and sort by average price per SQ Ft.
 
-    The houses dataset contains a collection of recent real estate listings in San Luis Obispo county and
-    around it. 
+    The houses dataset contains a collection of recent real estate listings in 
+    San Luis Obispo county and around it. 
 
     The dataset contains the following fields:
     1. MLS: Multiple listing service number for the house (unique ID).
-    2. Location: city/town where the house is located. Most locations are in San Luis Obispo county and
-    northern Santa Barbara county (Santa Maria­Orcutt, Lompoc, Guadelupe, Los Alamos), but there
-    some out of area locations as well.
+    2. Location: city/town where the house is located. Most locations are in 
+        San Luis Obispo county and northern Santa Barbara county (Santa Maria­Orcutt, Lompoc, 
+        Guadelupe, Los Alamos), but there some out of area locations as well.
     3. Price: the most recent listing price of the house (in dollars).
     4. Bedrooms: number of bedrooms.
     5. Bathrooms: number of bathrooms.
     6. Size: size of the house in square feet.
     7. Price/SQ.ft: price of the house per square foot.
-    8. Status: type of sale. Thee types are represented in the dataset: Short Sale, Foreclosure and Regular.
+    8. Status: type of sale. Thee types are represented in the dataset: Short Sale, 
+        Foreclosure and Regular.
 
     Each field is comma separated.
 
diff --git a/sparkSql/HousePriceSolution.py b/sparkSql/HousePriceSolution.py
@@ -4,8 +4,8 @@
 
 if __name__ == "__main__":
 
-    session = SparkSession.builder.appName("HousePriceSolution").master("local").getOrCreate()
-    session.sparkContext.setLogLevel("ERROR")
+    session = SparkSession.builder.appName("HousePriceSolution").master("local[*]").getOrCreate()
+    
     realEstate = session.read \
         .option("header","true") \
         .option("inferSchema", value=True) \
diff --git a/sparkSql/RddDataframeConversion.py b/sparkSql/RddDataframeConversion.py
diff --git a/sparkSql/StackOverFlowSurvey.py b/sparkSql/StackOverFlowSurvey.py
diff --git a/sparkSql/join/UkMakerSpaces.py b/sparkSql/join/UkMakerSpaces.py
