[Github]Apache Spark 2 for Beginners

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

Apache Spark 2 for Beginners

本帖隐藏的内容

https://github.com/PacktPublishing/Apache-Spark-2-for-Beginners

This is the code repository for Apache Spark 2 for Beginners, published by Packt. It contains all the supporting project files necessary to work through the book from start to finish. ##Instructions and Navigations All of the code is organized into folders. Each folder starts with a number followed by the application name. For example, Chapter02.


Detailed installation steps (software-wise)

The steps should be listed in a way that it prepares the system environment to be able to test the codes of the book. ###1. Apache Spark: a. Download Spark version mentioned in the table
b. Build Spark from source or use the binary download and follow the detailed instructions given in the pagehttp://spark.apache.org/docs/latest/building-spark.html
c. If building Spark from source, make sure that the R profile is also built and the instructions to do that is given in the link given inthe step b.
###2. Apache Kafka a. Download Kafka version mentioned in the table
b. The “quick start” section of the Kafka documentation gives the instructions to setup Kafka.http://kafka.apache.org/documentation.html#quickstart
c. Apart from the installation instructions, the topic creation and the other Kafka setup pre-requisites have been covered in detail in the chapter of the book

The code will look like the following:

Python 3.5.0 (v3.5.0:374f501f4567, Sep 12 2015, 11:00:19)[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin

Spark 2.0.0 or above is to be installed on at least a standalone machine to run the code samples and do further activities to learn more about the subject. For Spark Stream Processing, Kafka needs to be installed and configured as a message broker with its command line producer producing messages and the application developed using Spark as a consumer of those messages.

##Related Products

• Scala Design Patterns

• Machine Learning with Spark

• Quickstart Apache Axis2 ###Suggestions and Feedback Click here if you have any feedback or suggestions.

  
  

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏1 回帖

关键词：Apache Spark beginners beginner apache beginn

1. # coding: utf-8
   
2. 
   
3. # In[7]:
   
4. 
   
5. # Import all the required liraries
   
6. from pyspark.sql import Row
   
7. import matplotlib.pyplot as plt
   
8. import numpy as np
   
9. import matplotlib.pyplot as plt
   
10. import pylab as P
    
11. get_ipython().magic('matplotlib inline')
    
12. plt.rcdefaults()
    
13. 
    
14. 
    
15. # In[8]:
    
16. 
    
17. # TODO - The following location has to be changed to the appropriate data file location
    
18. dataDir = "/Users/RajT/Documents/Writing/SparkForBeginners/To-PACKTPUB/Contents/B05289-05-SparkDataAnalysisWithPython/Data/ml-100k/"
    
19. 
    
20. 
    
21. # In[9]:
    
22. 
    
23. # Create the DataFrame of the user data set
    
24. lines = sc.textFile(dataDir + "u.user")
    
25. splitLines = lines.map(lambda l: l.split("|"))
    
26. usersRDD = splitLines.map(lambda p: Row(id=p[0], age=int(p[1]), gender=p[2], occupation=p[3], zipcode=p[4]))
    
27. usersDF = spark.createDataFrame(usersRDD)
    
28. usersDF.createOrReplaceTempView("users")
    
29. usersDF.show()
    
30. 
    
31. 
    
32. # In[10]:
    
33. 
    
34. # Create the DataFrame of the user data set with only one column age
    
35. ageDF = spark.sql("SELECT age FROM users")
    
36. ageList = ageDF.rdd.map(lambda p: p.age).collect()
    
37. ageDF.describe().show()
    
38. 
    
39. 
    
40. # In[11]:
    
41. 
    
42. # Age distribution of the users
    
43. plt.hist(ageList)
    
44. plt.title("Age distribution of the users\n")
    
45. plt.xlabel("Age")
    
46. plt.ylabel("Number of users")
    
47. plt.show(block=False)
    
48. 
    
49. 
    
50. # In[12]:
    
51. 
    
52. # Draw a probability density plot
    
53. from scipy.stats import gaussian_kde
    
54. density = gaussian_kde(ageList)
    
55. xAxisValues = np.linspace(0,100,1000) # Use the range of ages from 0 to 100 and the number of data points
    
56. density.covariance_factor = lambda : .5
    
57. density._compute_covariance()
    
58. plt.title("Age density plot of the users\n")
    
59. plt.xlabel("Age")
    
60. plt.ylabel("Density")
    
61. plt.plot(xAxisValues, density(xAxisValues))
    
62. plt.show(block=False)
    
63. 
    
64. 
    
65. # In[13]:
    
66. 
    
67. # The following example demonstrates the creation of multiple diagrams in one figure
    
68. # There are two plots on one row
    
69. # The first one is the histogram of the distribution
    
70. # The second one is the boxplot containing the summary of the distribution
    
71. plt.subplot(121)
    
72. plt.hist(ageList)
    
73. plt.title("Age distribution of the users\n")
    
74. plt.xlabel("Age")
    
75. plt.ylabel("Number of users")
    
76. plt.subplot(122)
    
77. plt.title("Summary of distribution\n")
    
78. plt.xlabel("Age")
    
79. plt.boxplot(ageList, vert=False)
    
80. plt.show(block=False)
    
81. 
    
82. 
    
83. # In[14]:
    
84. 
    
85. occupationsTop10 = spark.sql("SELECT occupation, count(occupation) as usercount FROM users GROUP BY occupation ORDER BY usercount DESC LIMIT 10")
    
86. occupationsTop10.show()
    
87. 
    
88. 
    
89. # In[15]:
    
90. 
    
91. occupationsTop10Tuple = occupationsTop10.rdd.map(lambda p: (p.occupation,p.usercount)).collect()
    
92. occupationsTop10List, countTop10List = zip(*occupationsTop10Tuple)
    
93. occupationsTop10Tuple
    
94. 
    
95. 
    
96. # In[16]:
    
97. 
    
98. # Top 10 occupations in terms of the number of users having that occupation who have rated movies
    
99. y_pos = np.arange(len(occupationsTop10List))
    
100. plt.barh(y_pos, countTop10List, align='center', alpha=0.4)
     
101. plt.yticks(y_pos, occupationsTop10List)
     
102. plt.xlabel('Number of users')
     
103. plt.title('Top 10 user types\n')
     
104. plt.gcf().subplots_adjust(left=0.15)
     
105. plt.show(block=False)
     
106. 
     
107. 
     
108. # In[17]:
     
109. 
     
110. occupationsGender = spark.sql("SELECT occupation, gender FROM users")
     
111. occupationsGender.show()
     
112. 
     
113. 
     
114. # In[18]:
     
115. 
     
116. occCrossTab = occupationsGender.stat.crosstab("occupation", "gender")
     
117. occCrossTab.show()
     
118. 
     
119. 
     
120. # In[19]:
     
121. 
     
122. occCrossTab.describe('M', 'F').show()
     
123. 
     
124. 
     
125. # In[20]:
     
126. 
     
127. occCrossTab.stat.cov('M', 'F')
     
128. 
     
129. 
     
130. # In[21]:
     
131. 
     
132. occCrossTab.stat.corr('M', 'F')
     
133. 
     
134. 
     
135. # In[22]:
     
136. 
     
137. occupationsCrossTuple = occCrossTab.rdd.map(lambda p: (p.occupation_gender,p.M, p.F)).collect()
     
138. occList, mList, fList = zip(*occupationsCrossTuple)
     
139. 
     
140. 
     
141. # In[23]:
     
142. 
     
143. N = len(occList)
     
144. ind = np.arange(N)    # the x locations for the groups
     
145. width = 0.75          # the width of the bars: can also be len(x) sequence
     
146. p1 = plt.bar(ind, mList, width, color='r')
     
147. p2 = plt.bar(ind, fList, width, color='y', bottom=mList)
     
148. plt.ylabel('Count')
     
149. plt.title('Gender distribution by occupation\n')
     
150. plt.xticks(ind + width/2., occList, rotation=90)
     
151. plt.legend((p1[0], p2[0]), ('Male', 'Female'))
     
152. plt.gcf().subplots_adjust(bottom=0.25)
     
153. plt.show(block=False)
     
154. 
     
155. 
     
156. # In[24]:
     
157. 
     
158. occupationsBottom10 = spark.sql("SELECT occupation, count(occupation) as usercount FROM users GROUP BY occupation ORDER BY usercount LIMIT 10")
     
159. occupationsBottom10.show()
     
160. 
     
161. 
     
162. # In[25]:
     
163. 
     
164. occupationsBottom10Tuple = occupationsBottom10.rdd.map(lambda p: (p.occupation,p.usercount)).collect()
     
165. occupationsBottom10List, countBottom10List = zip(*occupationsBottom10Tuple)
     
166. 
     
167. 
     
168. # In[26]:
     
169. 
     
170. # Bottom 10 occupations in terms of the number of users having that occupation who have rated movies
     
171. explode = (0, 0, 0, 0,0.1,0,0,0,0,0.1)  
     
172. plt.pie(countBottom10List, explode=explode, labels=occupationsBottom10List, autopct='%1.1f%%', shadow=True, startangle=90)
     
173. plt.title('Bottom 10 user types\n')
     
174. plt.show(block=False)
     
175. 
     
176. 
     
177. # In[27]:
     
178. 
     
179. zipTop10 = spark.sql("SELECT zipcode, count(zipcode) as usercount FROM users GROUP BY zipcode ORDER BY usercount DESC LIMIT 10")
     
180. zipTop10.show()
     
181. 
     
182. 
     
183. # In[28]:
     
184. 
     
185. zipTop10Tuple = zipTop10.rdd.map(lambda p: (p.zipcode,p.usercount)).collect()
     
186. zipTop10List, countTop10List = zip(*zipTop10Tuple)
     
187. 
     
188. 
     
189. # In[29]:
     
190. 
     
191. # Top 10 zipcodes in terms of the number of users living in that zipcode who have rated movies
     
192. explode = (0.1, 0, 0, 0,0,0,0,0,0,0)  # explode a slice if required
     
193. plt.pie(countTop10List, explode=explode, labels=zipTop10List, autopct='%1.1f%%', shadow=True)
     
194. #draw a circle at the center of pie to make it look like a donut
     
195. centre_circle = plt.Circle((0,0),0.75,color='black', fc='white',linewidth=1.25)
     
196. plt.gcf().gca().add_artist(centre_circle)
     
197. # The aspect ratio is to be made equal. This is to make sure that pie chart is coming perfectly as a circle.
     
198. plt.axis('equal')
     
199. plt.text(- 0.25,0,'Top 10 zip codes')
     
200. plt.show(block=False)
     
201. 
     
202. 
     
203. # In[30]:
     
204. 
     
205. ages = spark.sql("SELECT occupation, age FROM users WHERE occupation ='administrator' ORDER BY age")
     
206. adminAges = ages.rdd.map(lambda p: p.age).collect()
     
207. ages.describe().show()
     
208. 
     
209. 
     
210. # In[31]:
     
211. 
     
212. ages = spark.sql("SELECT occupation, age FROM users WHERE occupation ='engineer' ORDER BY age")
     
213. engAges = ages.rdd.map(lambda p: p.age).collect()
     
214. ages.describe().show()
     
215. 
     
216. 
     
217. # In[32]:
     
218. 
     
219. ages = spark.sql("SELECT occupation, age FROM users WHERE occupation ='programmer' ORDER BY age")
     
220. progAges = ages.rdd.map(lambda p: p.age).collect()
     
221. ages.describe().show()
     
222. 
     
223. 
     
224. # In[33]:
     
225. 
     
226. # Box plots of the ages by profession
     
227. boxPlotAges = [adminAges, engAges, progAges]
     
228. boxPlotLabels = ['administrator','engineer', 'programmer' ]
     
229. x = np.arange(len(boxPlotLabels))
     
230. plt.figure()
     
231. plt.boxplot(boxPlotAges)
     
232. plt.title('Age summary statistics\n')
     
233. plt.ylabel("Age")
     
234. plt.xticks(x + 1, boxPlotLabels, rotation=0)
     
235. plt.show(block=False)
     
236. 
     
237. 
     
238. # In[34]:
     
239. 
     
240. movieLines = sc.textFile(dataDir + "u.item")
     
241. splitMovieLines = movieLines.map(lambda l: l.split("|"))
     
242. moviesRDD = splitMovieLines.map(lambda p: Row(id=p[0], title=p[1], releaseDate=p[2], videoReleaseDate=p[3], url=p[4], unknown=int(p[5]),action=int(p[6]),adventure=int(p[7]),animation=int(p[8]),childrens=int(p[9]),comedy=int(p[10]),crime=int(p[11]),documentary=int(p[12]),drama=int(p[13]),fantasy=int(p[14]),filmNoir=int(p[15]),horror=int(p[16]),musical=int(p[17]),mystery=int(p[18]),romance=int(p[19]),sciFi=int(p[20]),thriller=int(p[21]),war=int(p[22]),western=int(p[23])))
     
243. moviesDF = spark.createDataFrame(moviesRDD)
     
244. moviesDF.createOrReplaceTempView("movies")
     
245. 
     
246. 
     
247. # In[35]:
     
248. 
     
249. genreDF = spark.sql("SELECT sum(unknown) as unknown, sum(action) as action,sum(adventure) as adventure,sum(animation) as animation, sum(childrens) as childrens,sum(comedy) as comedy,sum(crime) as crime,sum(documentary) as documentary,sum(drama) as drama,sum(fantasy) as fantasy,sum(filmNoir) as filmNoir,sum(horror) as horror,sum(musical) as musical,sum(mystery) as mystery,sum(romance) as romance,sum(sciFi) as sciFi,sum(thriller) as thriller,sum(war) as war,sum(western) as western FROM movies")
     
250. genreList = genreDF.collect()
     
251. genreDict = genreList[0].asDict()
     
252. labelValues = list(genreDict.keys())
     
253. countList = list(genreDict.values())
     
254. genreDict
     
255. 
     
256. 
     
257. # In[36]:
     
258. 
     
259. # http://opentechschool.github.io/python-data-intro/core/charts.html
     
260. # Movie types and the counts
     
261. x = np.arange(len(labelValues))
     
262. plt.title('Movie types\n')
     
263. plt.ylabel("Count")
     
264. plt.bar(x, countList)
     
265. plt.xticks(x + 0.5, labelValues, rotation=90)
     
266. plt.gcf().subplots_adjust(bottom=0.20)
     
267. plt.show(block=False)
     
268. 
     
269. 
     
270. # In[37]:
     
271. 
     
272. yearDF = sqlContext.sql("SELECT substring(releaseDate,8,4) as releaseYear, count(*) as movieCount FROM movies GROUP BY substring(releaseDate,8,4) ORDER BY movieCount DESC LIMIT 10")
     
273. yearDF.show()
     
274. yearMovieCountTuple = yearDF.rdd.map(lambda p: (int(p.releaseYear),p.movieCount)).collect()
     
275. yearList,movieCountList = zip(*yearMovieCountTuple)
     
276. countArea = yearDF.rdd.map(lambda p: np.pi * (p.movieCount/15)**2).collect()
     
277. 
     
278. 
     
279. # In[38]:
     
280. 
     
281. # Top 10 years where the most number of movies have been released
     
282. plt.title('Top 10 movie release by year\n')
     
283. plt.xlabel("Year")
     
284. plt.ylabel("Number of movies released")
     
285. plt.ylim([0,max(movieCountList)])
     
286. colors = np.random.rand(10)
     
287. plt.scatter(yearList, movieCountList,c=colors)
     
288. plt.show(block=False)
     
289. 
     
290. 
     
291. # In[39]:
     
292. 
     
293. # Top 10 years where the most number of movies have been released
     
294. plt.title('Top 10 movie release by year\n')
     
295. plt.xlabel("Year")
     
296. plt.ylabel("Number of movies released")
     
297. plt.ylim([0,max(movieCountList) + 100])
     
298. colors = np.random.rand(10)
     
299. plt.scatter(yearList, movieCountList,c=colors, s=countArea)
     
300. plt.show(block=False)
     
301. 
     
302. 
     
303. # In[40]:
     
304. 
     
305. yearActionDF = spark.sql("SELECT substring(releaseDate,8,4) as actionReleaseYear, count(*) as actionMovieCount FROM movies WHERE action = 1 GROUP BY substring(releaseDate,8,4) ORDER BY actionReleaseYear DESC LIMIT 10")
     
306. yearActionDF.show()
     
307. yearActionDF.createOrReplaceTempView("action")
     
308. 
     
309. 
     
310. # In[41]:
     
311. 
     
312. yearDramaDF = spark.sql("SELECT substring(releaseDate,8,4) as dramaReleaseYear, count(*) as dramaMovieCount FROM movies WHERE drama = 1 GROUP BY substring(releaseDate,8,4) ORDER BY dramaReleaseYear DESC LIMIT 10")
     
313. yearDramaDF.show()
     
314. yearDramaDF.createOrReplaceTempView("drama")
     
315. 
     
316. 
     
317. # In[42]:
     
318. 
     
319. yearCombinedDF = spark.sql("SELECT a.actionReleaseYear as releaseYear, a.actionMovieCount, d.dramaMovieCount FROM action a, drama d WHERE a.actionReleaseYear = d.dramaReleaseYear ORDER BY a.actionReleaseYear DESC LIMIT 10")
     
320. yearCombinedDF.show()
     
321. 
     
322. 
     
323. # In[43]:
     
324. 
     
325. yearMovieCountTuple = yearCombinedDF.rdd.map(lambda p: (p.releaseYear,p.actionMovieCount, p.dramaMovieCount)).collect()
     
326. yearList,actionMovieCountList,dramaMovieCountList = zip(*yearMovieCountTuple)
     
327. 
     
328. 
     
329. # In[44]:
     
330. 
     
331. plt.title("Movie release by year\n")
     
332. plt.xlabel("Year")
     
333. plt.ylabel("Movie count")
     
334. line_action, = plt.plot(yearList, actionMovieCountList)
     
335. line_drama, = plt.plot(yearList, dramaMovieCountList)
     
336. plt.legend([line_action, line_drama], ['Action Movies', 'Drama Movies'],loc='upper left')
     
337. plt.gca().get_xaxis().get_major_formatter().set_useOffset(False)
     
338. plt.show(block=False)
     
339. 
     
340. 
     
341. # In[ ]:

复制代码

谢谢分享

谢谢楼主分享！

[Github]Apache Spark 2 for Beginners

谢谢分享

thank you sir