Merge pull request jleetutorial#3 from jleetutorial/pedromb-scala_to_python

Scala to Python - advanced folder

Author: jleetutorial

advanced/accumulator/StackOverFlowSurvey.py

@@ -0,0 +1,25 @@
+from pyspark import SparkContext
+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")
+
+    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))
+
+    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))

advanced/accumulator/StackOverFlowSurvey.scala

@@ -0,0 +1,28 @@
+from pyspark import SparkContext
+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")
+
+    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))
+
+    print("Count of responses from Canada: {}".format(responseFromCanada.count()))
+    print("Number of bytes processed: {}".format(processedBytes.value))
+    print("Total count of responses: {}".format(total.value))
+    print("Count of responses missing salary middle point: {}".format(missingSalaryMidPoint.value))

advanced/accumulator/StackOverFlowSurveyFollowUp.py

@@ -0,0 +1,29 @@
+from pyspark import SparkContext
+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}
+
+if __name__ == "__main__":
+    sc = SparkContext("local", "UkMakerSpaces")
+    sc.setLogLevel("ERROR")
+
+    postCodeMap = sc.broadcast(loadPostCodeMap())
+
+    makerSpaceRdd = sc.textFile("in/uk-makerspaces-identifiable-data.csv")
+
+    regions = makerSpaceRdd \
+      .filter(lambda line: Utils.COMMA_DELIMITER.split(line)[0] != "Timestamp") \
+      .filter(lambda line: getPostPrefix(line) is not None) \
+      .map(lambda line: postCodeMap.value[getPostPrefix(line)] \
+        if getPostPrefix(line) in postCodeMap.value else "Unknow")
+
+    for region, count in regions.countByValue().items():
+        print("{} : {}".format(region, count))

advanced/accumulator/StackOverFlowSurveyFollowUp.scala

@@ -0,0 +1,26 @@
+from pyspark import SparkContext
+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}
+
+if __name__ == "__main__":
+    sc = SparkContext("local", "UkMakerSpaces")
+    sc.setLogLevel("ERROR")
+    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"))
+
+    for region, count in regions.countByValue().items():
+        print("{} : {}".format(region, count))