[GitHub]Taming Big Data with Apache Spark and Python

About the Book

Frank Kane’s Taming Big Data with Apache Spark and Python is your companion to learning Apache Spark in a hands-on manner. Frank will start you off by teaching you how to set up Spark on a single system or on a cluster, and you’ll soon move on to analyzing large data sets using Spark RDD, and developing and running effective Spark jobs quickly using Python.

Apache Spark has emerged as the next big thing in the Big Data domain – quickly rising from an ascending technology to an established superstar in just a matter of years. Spark allows you to quickly extract actionable insights from large amounts of data, on a real-time basis, making it an essential tool in many modern businesses.

Frank has packed this book with over 15 interactive, fun-filled examples relevant to the real world, and he will empower you to understand the Spark ecosystem and implement production-grade real-time Spark projects with ease. ##Instructions and Navigation All of the code is organized into folders. Each folder starts with a number followed by the application name. For example, Chapter02.

The code will look like the following:

from pyspark import SparkConf, SparkContext import collections conf = SparkConf().setMaster("local").setAppName("RatingsHistogram") sc = SparkContext(conf = conf)lines = sc.textFile("file:///SparkCourse/ml-100k/u.data") ratings = lines.map(lambda x: x.split()[2]) result = ratings.countByValue() sortedResults = collections.OrderedDict(sorted(result.items())) for key, value in sortedResults.items():     print("%s %i" % (key, value))

For this book you’ll need a Python development environment (Python 3.5 or newer), a Canopy installer, Java Development Kit, and of course Spark itself (Spark 2.0 and beyond).

We'll show you how to install this software in first chapter of the book.

This book is based on the Windows operating system, so installations are provided according to it. If you have Mac or Linux, you can follow this URL http://media.sundog-soft.com/spark-python-install.pdf, which contains written instructions on getting everything set up on Mac OS and on Linux.

Related Products

• Taming Big Data with Spark Streaming and Scala – Hands On! [Video]

• Taming Big Data with Apache Spark and Python - Hands On! [Video]

• Taming Big Data with MapReduce and Hadoop - Hands On! [Video]

  
  

Suggestions and Feedback

Click here if you have any feedback or suggestions.

1. from pyspark import SparkConf, SparkContext
   
2. 
   
3. conf = SparkConf().setMaster("local").setAppName("FriendsByAge")
   
4. sc = SparkContext(conf = conf)
   
5. 
   
6. def parseLine(line):
   
7.     fields = line.split(',')
   
8.     age = int(fields[2])
   
9.     numFriends = int(fields[3])
   
10.     return (age, numFriends)
    
11. 
    
12. lines = sc.textFile("file:///SparkCourse/fakefriends.csv")
    
13. rdd = lines.map(parseLine)
    
14. totalsByAge = rdd.mapValues(lambda x: (x, 1)).reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))
    
15. averagesByAge = totalsByAge.mapValues(lambda x: x[0] / x[1])
    
16. results = averagesByAge.collect()
    
17. for result in results:
    
18.     print(result)

复制代码

1. from pyspark import SparkConf, SparkContext
   
2. 
   
3. conf = SparkConf().setMaster("local").setAppName("MaxTemperatures")
   
4. sc = SparkContext(conf = conf)
   
5. 
   
6. def parseLine(line):
   
7.     fields = line.split(',')
   
8.     stationID = fields[0]
   
9.     entryType = fields[2]
   
10.     temperature = float(fields[3]) * 0.1 * (9.0 / 5.0) + 32.0
    
11.     return (stationID, entryType, temperature)
    
12. 
    
13. lines = sc.textFile("file:///SparkCourse/1800.csv")
    
14. parsedLines = lines.map(parseLine)
    
15. maxTemps = parsedLines.filter(lambda x: "TMAX" in x[1])
    
16. stationTemps = maxTemps.map(lambda x: (x[0], x[2]))
    
17. maxTemps = stationTemps.reduceByKey(lambda x, y: max(x,y))
    
18. results = maxTemps.collect();
    
19. 
    
20. for result in results:
    
21.     print(result[0] + "\t{:.2f}F".format(result[1]))

复制代码

1. from pyspark import SparkConf, SparkContext
   
2. 
   
3. conf = SparkConf().setMaster("local").setAppName("MinTemperatures")
   
4. sc = SparkContext(conf = conf)
   
5. 
   
6. def parseLine(line):
   
7.     fields = line.split(',')
   
8.     stationID = fields[0]
   
9.     entryType = fields[2]
   
10.     temperature = float(fields[3]) * 0.1 * (9.0 / 5.0) + 32.0
    
11.     return (stationID, entryType, temperature)
    
12. 
    
13. lines = sc.textFile("file:///SparkCourse/1800.csv")
    
14. parsedLines = lines.map(parseLine)
    
15. minTemps = parsedLines.filter(lambda x: "TMIN" in x[1])
    
16. stationTemps = minTemps.map(lambda x: (x[0], x[2]))
    
17. minTemps = stationTemps.reduceByKey(lambda x, y: min(x,y))
    
18. results = minTemps.collect();
    
19. 
    
20. for result in results:
    
21.     print(result[0] + "\t{:.2f}F".format(result[1]))

复制代码

1. from pyspark import SparkConf, SparkContext
   
2. 
   
3. conf = SparkConf().setMaster("local").setAppName("PopularHero")
   
4. sc = SparkContext(conf = conf)
   
5. 
   
6. def countCoOccurences(line):
   
7.     elements = line.split()
   
8.     return (int(elements[0]), len(elements) - 1)
   
9. 
   
10. def parseNames(line):
    
11.     fields = line.split('\"')
    
12.     return (int(fields[0]), fields[1].encode("utf8"))
    
13. 
    
14. names = sc.textFile("file:///SparkCourse/marvel-names.txt")
    
15. namesRdd = names.map(parseNames)
    
16. 
    
17. lines = sc.textFile("file:///SparkCourse/marvel-graph.txt")
    
18. 
    
19. pairings = lines.map(countCoOccurences)
    
20. totalFriendsByCharacter = pairings.reduceByKey(lambda x, y : x + y)
    
21. flipped = totalFriendsByCharacter.map(lambda (x,y) : (y,x))
    
22. 
    
23. mostPopular = flipped.max()
    
24. 
    
25. mostPopularName = namesRdd.lookup(mostPopular[1])[0]
    
26. 
    
27. print(mostPopularName + " is the most popular superhero, with " + \
    
28.     str(mostPopular[0]) + " co-appearances.")

复制代码

1. import sys
   
2. from pyspark import SparkConf, SparkContext
   
3. from pyspark.mllib.recommendation import ALS, Rating
   
4. 
   
5. def loadMovieNames():
   
6.     movieNames = {}
   
7.     with open("ml-1m/movies.dat") as f:
   
8.         for line in f:
   
9.             fields = line.split('::')
   
10.             movieNames[int(fields[0])] = fields[1].decode('ascii', 'ignore')
    
11.     return movieNames
    
12. 
    
13. conf = SparkConf().setMaster("local[*]").setAppName("MovieRecommendationsALS")
    
14. sc = SparkContext(conf = conf)
    
15. sc.setCheckpointDir('checkpoint')
    
16. 
    
17. print("\nLoading movie names...")
    
18. nameDict = loadMovieNames()
    
19. 
    
20. data = sc.textFile("file:///E:/SparkCourse/ml-1m/ratings.dat")
    
21. 
    
22. ratings = data.map(lambda l: l.split("::")).map(lambda l: Rating(int(l[0]), int(l[1]), float(l[2]))).cache()
    
23. 
    
24. # Build the recommendation model using Alternating Least Squares
    
25. print("\nTraining recommendation model...")
    
26. rank = 10
    
27. numIterations = 20
    
28. model = ALS.train(ratings, rank, numIterations)
    
29. 
    
30. userID = int(sys.argv[1])
    
31. 
    
32. print("\nRatings for user ID " + str(userID) + ":")
    
33. userRatings = ratings.filter(lambda l: l[0] == userID)
    
34. for rating in userRatings.collect():
    
35.     print(nameDict[int(rating[1])] + ": " + str(rating[2]))
    
36. 
    
37. print("\nTop 10 recommendations:")
    
38. recommendations = model.recommendProducts(userID, 10)
    
39. for recommendation in recommendations:
    
40.     print(nameDict[int(recommendation[1])] + \
    
41.         " score " + str(recommendation[2]))

复制代码

1. import sys
   
2. from pyspark import SparkConf, SparkContext
   
3. from math import sqrt
   
4. 
   
5. #To run on EMR successfully + output results for Star Wars:
   
6. #aws s3 cp s3://sundog-spark/MovieSimilarities1M.py ./
   
7. #aws s3 sp c3://sundog-spark/ml-1m/movies.dat ./
   
8. #spark-submit --executor-memory 1g MovieSimilarities1M.py 260
   
9. 
   
10. def loadMovieNames():
    
11.     movieNames = {}
    
12.     with open("movies.dat") as f:
    
13.         for line in f:
    
14.             fields = line.split("::")
    
15.             movieNames[int(fields[0])] = fields[1].decode('ascii', 'ignore')
    
16.     return movieNames
    
17. 
    
18. def makePairs((user, ratings)):
    
19.     (movie1, rating1) = ratings[0]
    
20.     (movie2, rating2) = ratings[1]
    
21.     return ((movie1, movie2), (rating1, rating2))
    
22. 
    
23. def filterDuplicates( (userID, ratings) ):
    
24.     (movie1, rating1) = ratings[0]
    
25.     (movie2, rating2) = ratings[1]
    
26.     return movie1 < movie2
    
27. 
    
28. def computeCosineSimilarity(ratingPairs):
    
29.     numPairs = 0
    
30.     sum_xx = sum_yy = sum_xy = 0
    
31.     for ratingX, ratingY in ratingPairs:
    
32.         sum_xx += ratingX * ratingX
    
33.         sum_yy += ratingY * ratingY
    
34.         sum_xy += ratingX * ratingY
    
35.         numPairs += 1
    
36. 
    
37.     numerator = sum_xy
    
38.     denominator = sqrt(sum_xx) * sqrt(sum_yy)
    
39. 
    
40.     score = 0
    
41.     if (denominator):
    
42.         score = (numerator / (float(denominator)))
    
43. 
    
44.     return (score, numPairs)
    
45. 
    
46. 
    
47. conf = SparkConf()
    
48. sc = SparkContext(conf = conf)
    
49. 
    
50. print("\nLoading movie names...")
    
51. nameDict = loadMovieNames()
    
52. 
    
53. data = sc.textFile("s3n://sundog-spark/ml-1m/ratings.dat")
    
54. 
    
55. # Map ratings to key / value pairs: user ID => movie ID, rating
    
56. ratings = data.map(lambda l: l.split("::")).map(lambda l: (int(l[0]), (int(l[1]), float(l[2]))))
    
57. 
    
58. # Emit every movie rated together by the same user.
    
59. # Self-join to find every combination.
    
60. ratingsPartitioned = ratings.partitionBy(100)
    
61. joinedRatings = ratingsPartitioned.join(ratingsPartitioned)
    
62. 
    
63. # At this point our RDD consists of userID => ((movieID, rating), (movieID, rating))
    
64. 
    
65. # Filter out duplicate pairs
    
66. uniqueJoinedRatings = joinedRatings.filter(filterDuplicates)
    
67. 
    
68. # Now key by (movie1, movie2) pairs.
    
69. moviePairs = uniqueJoinedRatings.map(makePairs).partitionBy(100)
    
70. 
    
71. # We now have (movie1, movie2) => (rating1, rating2)
    
72. # Now collect all ratings for each movie pair and compute similarity
    
73. moviePairRatings = moviePairs.groupByKey()
    
74. 
    
75. # We now have (movie1, movie2) = > (rating1, rating2), (rating1, rating2) ...
    
76. # Can now compute similarities.
    
77. moviePairSimilarities = moviePairRatings.mapValues(computeCosineSimilarity).persist()
    
78. 
    
79. # Save the results if desired
    
80. moviePairSimilarities.sortByKey()
    
81. moviePairSimilarities.saveAsTextFile("movie-sims")
    
82. 
    
83. # Extract similarities for the movie we care about that are "good".
    
84. if (len(sys.argv) > 1):
    
85. 
    
86.     scoreThreshold = 0.97
    
87.     coOccurenceThreshold = 1000
    
88. 
    
89.     movieID = int(sys.argv[1])
    
90. 
    
91.     # Filter for movies with this sim that are "good" as defined by
    
92.     # our quality thresholds above
    
93.     filteredResults = moviePairSimilarities.filter(lambda((pair,sim)): \
    
94.         (pair[0] == movieID or pair[1] == movieID) \
    
95.         and sim[0] > scoreThreshold and sim[1] > coOccurenceThreshold)
    
96. 
    
97.     # Sort by quality score.
    
98.     results = filteredResults.map(lambda((pair,sim)): (sim, pair)).sortByKey(ascending = False).take(10)
    
99. 
    
100.     print("Top 10 similar movies for " + nameDict[movieID])
     
101.     for result in results:
     
102.         (sim, pair) = result
     
103.         # Display the similarity result that isn't the movie we're looking at
     
104.         similarMovieID = pair[0]
     
105.         if (similarMovieID == movieID):
     
106.             similarMovieID = pair[1]
     
107.         print(nameDict[similarMovieID] + "\tscore: " + str(sim[0]) + "\tstrength: "

复制代码

1. from __future__ import print_function
   
2. 
   
3. from pyspark.ml.regression import LinearRegression
   
4. 
   
5. from pyspark.sql import SparkSession
   
6. from pyspark.ml.linalg import Vectors
   
7. 
   
8. if __name__ == "__main__":
   
9. 
   
10.     # Create a SparkSession (Note, the config section is only for Windows!)
    
11.     spark = SparkSession.builder.config("spark.sql.warehouse.dir", "file:///C:/temp").appName("LinearRegression").getOrCreate()
    
12. 
    
13.     # Load up our data and convert it to the format MLLib expects.
    
14.     inputLines = spark.sparkContext.textFile("regression.txt")
    
15.     data = inputLines.map(lambda x: x.split(",")).map(lambda x: (float(x[0]), Vectors.dense(float(x[1]))))
    
16. 
    
17.     # Convert this RDD to a DataFrame
    
18.     colNames = ["label", "features"]
    
19.     df = data.toDF(colNames)
    
20. 
    
21.     # Note, there are lots of cases where you can avoid going from an RDD to a DataFrame.
    
22.     # Perhaps you're importing data from a real database. Or you are using structured streaming
    
23.     # to get your data.
    
24. 
    
25.     # Let's split our data into training data and testing data
    
26.     trainTest = df.randomSplit([0.5, 0.5])
    
27.     trainingDF = trainTest[0]
    
28.     testDF = trainTest[1]
    
29. 
    
30.     # Now create our linear regression model
    
31.     lir = LinearRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)
    
32. 
    
33.     # Train the model using our training data
    
34.     model = lir.fit(trainingDF)
    
35. 
    
36.     # Now see if we can predict values in our test data.
    
37.     # Generate predictions using our linear regression model for all features in our
    
38.     # test dataframe:
    
39.     fullPredictions = model.transform(testDF).cache()
    
40. 
    
41.     # Extract the predictions and the "known" correct labels.
    
42.     predictions = fullPredictions.select("prediction").rdd.map(lambda x: x[0])
    
43.     labels = fullPredictions.select("label").rdd.map(lambda x: x[0])
    
44. 
    
45.     # Zip them together
    
46.     predictionAndLabel = predictions.zip(labels).collect()
    
47. 
    
48.     # Print out the predicted and actual values for each point
    
49.     for prediction in predictionAndLabel:
    
50.       print(prediction)
    
51. 
    
52. 
    
53.     # Stop the session
    
54.     spark.stop()

复制代码

1. from pyspark.sql import SparkSession
   
2. from pyspark.sql import Row
   
3. 
   
4. import collections
   
5. 
   
6. # Create a SparkSession (Note, the config section is only for Windows!)
   
7. spark = SparkSession.builder.config("spark.sql.warehouse.dir", "file:///C:/temp").appName("SparkSQL").getOrCreate()
   
8. 
   
9. def mapper(line):
   
10.     fields = line.split(',')
    
11.     return Row(ID=int(fields[0]), name=str(fields[1].encode("utf-8")), age=int(fields[2]), numFriends=int(fields[3]))
    
12. 
    
13. lines = spark.sparkContext.textFile("fakefriends.csv")
    
14. people = lines.map(mapper)
    
15. 
    
16. # Infer the schema, and register the DataFrame as a table.
    
17. schemaPeople = spark.createDataFrame(people).cache()
    
18. schemaPeople.createOrReplaceTempView("people")
    
19. 
    
20. # SQL can be run over DataFrames that have been registered as a table.
    
21. teenagers = spark.sql("SELECT * FROM people WHERE age >= 13 AND age <= 19")
    
22. 
    
23. # The results of SQL queries are RDDs and support all the normal RDD operations.
    
24. for teen in teenagers.collect():
    
25.   print(teen)
    
26. 
    
27. # We can also use functions instead of SQL queries:
    
28. schemaPeople.groupBy("age").count().orderBy("age").show()
    
29. 
    
30. spark.stop()

复制代码

1. from pyspark import SparkConf, SparkContext
   
2. 
   
3. conf = SparkConf().setMaster("local").setAppName("WordCount")
   
4. sc = SparkContext(conf = conf)
   
5. 
   
6. input = sc.textFile("file:///sparkcourse/book.txt")
   
7. words = input.flatMap(lambda x: x.split())
   
8. wordCounts = words.countByValue()
   
9. 
   
10. for word, count in wordCounts.items():
    
11.     cleanWord = word.encode('ascii', 'ignore')
    
12.     if (cleanWord):
    
13.         print(cleanWord.decode() + " " + str(count))

复制代码