Oct 2, 2017 · Oct 2, 2017 · Oct 2, 2017 · Oct 2, 2017 · Oct 2, 2017 · Oct 2, 2017
diff --git a/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.py b/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    sc = SparkContext("local", "AverageHousePrice")
    sc.setLogLevel("ERROR")

    lines = sc.textFile("in/RealEstate.csv")
    cleanedLines = lines.filter(lambda line: "Bedrooms" not in line)

    housePricePairRdd = cleanedLines.map(lambda line: (line.split(",")[3], float(line.split(",")[2])))

    createCombiner = lambda x: (1, x)
    mergeValue = lambda avgCount, x: (avgCount[0] + 1, avgCount[1] + x)
    mergeCombiners = lambda avgCountA, avgCountB: (avgCountA[0] + avgCountB[0], avgCountA[1] + avgCountB[1])

    housePriceTotal = housePricePairRdd.combineByKey(createCombiner, mergeValue, mergeCombiners)

    housePriceAvg = housePriceTotal.mapValues(lambda avgCount: avgCount[1] / avgCount[0])
    for bedrooms, avgPrice in housePriceAvg.collect():
        print("{} : {}".format(bedrooms, avgPrice))
diff --git a/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.scala b/pairRdd/aggregation/combinebykey/AverageHousePriceSolution.scala
diff --git a/pairRdd/aggregation/reducebykey/WordCount.py b/pairRdd/aggregation/reducebykey/WordCount.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    sc = SparkContext("local", "wordCounts")
    sc.setLogLevel("ERROR")

    lines = sc.textFile("in/word_count.text")
    wordRdd = lines.flatMap(lambda line: line.split(" "))
    wordPairRdd = wordRdd.map(lambda word: (word, 1))

    wordCounts = wordPairRdd.reduceByKey(lambda x, y: x + y)
    for word, count in wordCounts.collect():
        print("{} : {}".format(word, count))
diff --git a/pairRdd/aggregation/reducebykey/WordCount.scala b/pairRdd/aggregation/reducebykey/WordCount.scala
diff --git a/...housePrice/AverageHousePriceProblem.scala → ...ey/housePrice/AverageHousePriceProblem.py b/...housePrice/AverageHousePriceProblem.scala → ...ey/housePrice/AverageHousePriceProblem.py
 package com.sparkTutorial.pairRdd.aggregation.reducebykey.housePrice
 from pyspark import SparkContext

 object AverageHousePriceProblem {
 if __name__ == "__main__":

  def main(args: Array[String]) {

    /* Create a Spark program to read the house data from in/RealEstate.csv,
       output the average price for houses with different number of bedrooms.
    '''
    Create a Spark program to read the house data from in/RealEstate.csv,
    output the average price for houses with different number of bedrooms.

    The houses dataset contains a collection of recent real estate listings in San Luis Obispo county and
    around it. 
       (2, 325000)
       ...

       3, 1 and 2 mean the number of bedrooms. 325000 means the average price of houses with 3 bedrooms is 325000.
     */
  }
    3, 1 and 2 mean the number of bedrooms. 325000 means the average price of houses with 3 bedrooms is 325000.

 }
    '''
diff --git a/pairRdd/aggregation/reducebykey/housePrice/AverageHousePriceSolution.py b/pairRdd/aggregation/reducebykey/housePrice/AverageHousePriceSolution.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    sc = SparkContext("local", "avgHousePrice")
    sc.setLogLevel("ERROR")

    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]))))

    housePriceTotal = housePricePairRdd \
        .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))

    print("housePriceTotal: ")
    for bedroom, total in housePriceTotal.collect():
        print("{} : {}".format(bedroom, total))

    housePriceAvg = housePriceTotal.mapValues(lambda avgCount: avgCount[1] / avgCount[0])
    print("\nhousePriceAvg: ")
    for bedroom, avg in housePriceAvg.collect():
        print("{} : {}".format(bedroom, avg))
diff --git a/pairRdd/aggregation/reducebykey/housePrice/AverageHousePriceSolution.scala b/pairRdd/aggregation/reducebykey/housePrice/AverageHousePriceSolution.scala
diff --git a/pairRdd/aggregation/reducebykey/housePrice/AvgCount.py b/pairRdd/aggregation/reducebykey/housePrice/AvgCount.py
 class AvgCount():

    def __init__(self, count: int, total: float):
        self.count = count
        self.total = total


diff --git a/pairRdd/aggregation/reducebykey/housePrice/AvgCount.scala b/pairRdd/aggregation/reducebykey/housePrice/AvgCount.scala
diff --git a/pairRdd/create/PairRddFromRegularRdd.py b/pairRdd/create/PairRddFromRegularRdd.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    sc = SparkContext("local", "create")
    sc.setLogLevel("ERROR")

    inputStrings = ["Lily 23", "Jack 29", "Mary 29", "James 8"]
    regularRDDs = sc.parallelize(inputStrings)

    pairRDD = regularRDDs.map(lambda s: (s.split(" ")[0], s.split(" ")[1]))
    pairRDD.coalesce(1).saveAsTextFile("out/pair_rdd_from_regular_rdd")
diff --git a/pairRdd/create/PairRddFromRegularRdd.scala b/pairRdd/create/PairRddFromRegularRdd.scala
diff --git a/pairRdd/create/PairRddFromTupleList.py b/pairRdd/create/PairRddFromTupleList.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    sc = SparkContext("local", "create")
    sc.setLogLevel("ERROR")

    tuples = [("Lily", 23), ("Jack", 29), ("Mary", 29), ("James", 8)]
    pairRDD = sc.parallelize(tuples)

    pairRDD.coalesce(1).saveAsTextFile("out/pair_rdd_from_tuple_list")
diff --git a/pairRdd/create/PairRddFromTupleList.scala b/pairRdd/create/PairRddFromTupleList.scala
diff --git a/pairRdd/filter/AirportsNotInUsaProblem.py b/pairRdd/filter/AirportsNotInUsaProblem.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    '''
    Create a Spark program to read the airport data from in/airports.text;
    generate a pair RDD with airport name being the key and country name being the value.
    Then remove all the airports which are located in United States and output the pair RDD to out/airports_not_in_usa_pair_rdd.text

    Each row of the input file contains the following columns:
    Airport ID, Name of airport, Main city served by airport, Country where airport is located,
    IATA/FAA code, ICAO Code, Latitude, Longitude, Altitude, Timezone, DST, Timezone in Olson format

    Sample output:

    ("Kamloops", "Canada")
    ("Wewak Intl", "Papua New Guinea")
    ...

    '''
diff --git a/pairRdd/filter/AirportsNotInUsaProblem.scala b/pairRdd/filter/AirportsNotInUsaProblem.scala
diff --git a/pairRdd/filter/AirportsNotInUsaSolution.py b/pairRdd/filter/AirportsNotInUsaSolution.py
 from pyspark import SparkContext
 from commons.Utils import Utils

 if __name__ == "__main__":

    sc = SparkContext("local", "airports")
    sc.setLogLevel("ERROR")

    airportsRDD = sc.textFile("in/airports.text")

    airportPairRDD = airportsRDD.map(lambda line: \
        (Utils.COMMA_DELIMITER.split(line)[1],
         Utils.COMMA_DELIMITER.split(line)[3]))
    airportsNotInUSA = airportPairRDD.filter(lambda keyValue: keyValue[1] != "\"United States\"")

    airportsNotInUSA.saveAsTextFile("out/airports_not_in_usa_pair_rdd.text")
diff --git a/pairRdd/filter/AirportsNotInUsaSolution.scala b/pairRdd/filter/AirportsNotInUsaSolution.scala
diff --git a/pairRdd/groupbykey/AirportsByCountryProblem.py b/pairRdd/groupbykey/AirportsByCountryProblem.py
 from pyspark import SparkContext

 if __name__ == "__main__":

    '''
    Create a Spark program to read the airport data from in/airports.text,
    output the the list of the names of the airports located in each country.

    Each row of the input file contains the following columns:
    Airport ID, Name of airport, Main city served by airport, Country where airport is located, IATA/FAA code,
    ICAO Code, Latitude, Longitude, Altitude, Timezone, DST, Timezone in Olson format

    Sample output:

    "Canada", ["Bagotville", "Montreal", "Coronation", ...]
    "Norway" : ["Vigra", "Andenes", "Alta", "Bomoen", "Bronnoy",..]
    "Papua New Guinea",  ["Goroka", "Madang", ...]
    ...

    '''



diff --git a/pairRdd/groupbykey/AirportsByCountryProblem.scala b/pairRdd/groupbykey/AirportsByCountryProblem.scala
diff --git a/pairRdd/groupbykey/AirportsByCountrySolution.py b/pairRdd/groupbykey/AirportsByCountrySolution.py
 from pyspark import SparkContext
 from commons.Utils import Utils

 if __name__ == "__main__":

    sc = SparkContext("local", "airports")
    sc.setLogLevel("ERROR")

    lines = sc.textFile("in/airports.text")

    countryAndAirportNameAndPair = lines.map(lambda airport:\
         (Utils.COMMA_DELIMITER.split(airport)[3],
          Utils.COMMA_DELIMITER.split(airport)[1]))

    airportsByCountry = countryAndAirportNameAndPair.groupByKey()

    for country, airportName in airportsByCountry.collectAsMap().items():
        print("{}: {}".format(country,list(airportName)))
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,21 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		sc = SparkContext("local", "AverageHousePrice")
		sc.setLogLevel("ERROR")

		lines = sc.textFile("in/RealEstate.csv")
		cleanedLines = lines.filter(lambda line: "Bedrooms" not in line)

		housePricePairRdd = cleanedLines.map(lambda line: (line.split(",")[3], float(line.split(",")[2])))

		createCombiner = lambda x: (1, x)
		mergeValue = lambda avgCount, x: (avgCount[0] + 1, avgCount[1] + x)
		mergeCombiners = lambda avgCountA, avgCountB: (avgCountA[0] + avgCountB[0], avgCountA[1] + avgCountB[1])

		housePriceTotal = housePricePairRdd.combineByKey(createCombiner, mergeValue, mergeCombiners)

		housePriceAvg = housePriceTotal.mapValues(lambda avgCount: avgCount[1] / avgCount[0])
		for bedrooms, avgPrice in housePriceAvg.collect():
		print("{} : {}".format(bedrooms, avgPrice))
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,14 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		sc = SparkContext("local", "wordCounts")
		sc.setLogLevel("ERROR")

		lines = sc.textFile("in/word_count.text")
		wordRdd = lines.flatMap(lambda line: line.split(" "))
		wordPairRdd = wordRdd.map(lambda word: (word, 1))

		wordCounts = wordPairRdd.reduceByKey(lambda x, y: x + y)
		for word, count in wordCounts.collect():
		print("{} : {}".format(word, count))
Original file line number	Diff line number	Diff line change
		@@ -1,11 +1,10 @@
		package com.sparkTutorial.pairRdd.aggregation.reducebykey.housePrice
		from pyspark import SparkContext

		object AverageHousePriceProblem {
		if __name__ == "__main__":

		def main(args: Array[String]) {

		/* Create a Spark program to read the house data from in/RealEstate.csv,
		output the average price for houses with different number of bedrooms.
		'''
		Create a Spark program to read the house data from in/RealEstate.csv,
		output the average price for houses with different number of bedrooms.

		The houses dataset contains a collection of recent real estate listings in San Luis Obispo county and
		around it.
Expand All		@@ -31,8 +30,6 @@ def main(args: Array[String]) {
		(2, 325000)
		...

		3, 1 and 2 mean the number of bedrooms. 325000 means the average price of houses with 3 bedrooms is 325000.
		*/
		}
		3, 1 and 2 mean the number of bedrooms. 325000 means the average price of houses with 3 bedrooms is 325000.

		}
		'''
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,24 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		sc = SparkContext("local", "avgHousePrice")
		sc.setLogLevel("ERROR")

		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]))))

		housePriceTotal = housePricePairRdd \
		.reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))

		print("housePriceTotal: ")
		for bedroom, total in housePriceTotal.collect():
		print("{} : {}".format(bedroom, total))

		housePriceAvg = housePriceTotal.mapValues(lambda avgCount: avgCount[1] / avgCount[0])
		print("\nhousePriceAvg: ")
		for bedroom, avg in housePriceAvg.collect():
		print("{} : {}".format(bedroom, avg))
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,7 @@
		class AvgCount():

		def __init__(self, count: int, total: float):
		self.count = count
		self.total = total
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,12 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		sc = SparkContext("local", "create")
		sc.setLogLevel("ERROR")

		inputStrings = ["Lily 23", "Jack 29", "Mary 29", "James 8"]
		regularRDDs = sc.parallelize(inputStrings)

		pairRDD = regularRDDs.map(lambda s: (s.split(" ")[0], s.split(" ")[1]))
		pairRDD.coalesce(1).saveAsTextFile("out/pair_rdd_from_regular_rdd")
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,11 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		sc = SparkContext("local", "create")
		sc.setLogLevel("ERROR")

		tuples = [("Lily", 23), ("Jack", 29), ("Mary", 29), ("James", 8)]
		pairRDD = sc.parallelize(tuples)

		pairRDD.coalesce(1).saveAsTextFile("out/pair_rdd_from_tuple_list")
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,20 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		'''
		Create a Spark program to read the airport data from in/airports.text;
		generate a pair RDD with airport name being the key and country name being the value.
		Then remove all the airports which are located in United States and output the pair RDD to out/airports_not_in_usa_pair_rdd.text

		Each row of the input file contains the following columns:
		Airport ID, Name of airport, Main city served by airport, Country where airport is located,
		IATA/FAA code, ICAO Code, Latitude, Longitude, Altitude, Timezone, DST, Timezone in Olson format

		Sample output:

		("Kamloops", "Canada")
		("Wewak Intl", "Papua New Guinea")
		...

		'''
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,16 @@
		from pyspark import SparkContext
		from commons.Utils import Utils

		if __name__ == "__main__":

		sc = SparkContext("local", "airports")
		sc.setLogLevel("ERROR")

		airportsRDD = sc.textFile("in/airports.text")

		airportPairRDD = airportsRDD.map(lambda line: \
		(Utils.COMMA_DELIMITER.split(line)[1],
		Utils.COMMA_DELIMITER.split(line)[3]))
		airportsNotInUSA = airportPairRDD.filter(lambda keyValue: keyValue[1] != "\"United States\"")

		airportsNotInUSA.saveAsTextFile("out/airports_not_in_usa_pair_rdd.text")
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,23 @@
		from pyspark import SparkContext

		if __name__ == "__main__":

		'''
		Create a Spark program to read the airport data from in/airports.text,
		output the the list of the names of the airports located in each country.

		Each row of the input file contains the following columns:
		Airport ID, Name of airport, Main city served by airport, Country where airport is located, IATA/FAA code,
		ICAO Code, Latitude, Longitude, Altitude, Timezone, DST, Timezone in Olson format

		Sample output:

		"Canada", ["Bagotville", "Montreal", "Coronation", ...]
		"Norway" : ["Vigra", "Andenes", "Alta", "Bomoen", "Bronnoy",..]
		"Papua New Guinea", ["Goroka", "Madang", ...]
		...

		'''