forked from aroegies/bigdata-2019w
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathassignment7-451.html
393 lines (308 loc) · 15.4 KB
/
assignment7-451.html
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1">
<!-- The above 3 meta tags *must* come first in the head; any other head content must come *after* these tags -->
<meta name="description" content="Course homepage for CS 431/631 451/651 Data-Intensive Distributed Computing (Winter 2019) at the University of Waterloo">
<meta name="author" content="Adam Roegiest">
<title>Data-Intensive Distributed Computing</title>
<!-- Bootstrap core CSS -->
<link href="css/bootstrap.min.css" rel="stylesheet">
<!-- IE10 viewport hack for Surface/desktop Windows 8 bug -->
<link href="css/ie10-viewport-bug-workaround.css" rel="stylesheet">
<!-- Just for debugging purposes. Don't actually copy these 2 lines! -->
<!--[if lt IE 9]><script src="../../assets/js/ie8-responsive-file-warning.js"></script><![endif]-->
<script src="js/ie-emulation-modes-warning.js"></script>
<!-- HTML5 shim and Respond.js for IE8 support of HTML5 elements and media queries -->
<!--[if lt IE 9]>
<script src="https://oss.maxcdn.com/html5shiv/3.7.3/html5shiv.min.js"></script>
<script src="https://oss.maxcdn.com/respond/1.4.2/respond.min.js"></script>
<![endif]-->
<style>
body {
padding-top: 60px; /* 60px to make the container go all the way to the bottom of the topbar */
}
</style>
</head>
<body>
<nav class="navbar navbar-inverse navbar-fixed-top">
<div class="container">
<div class="navbar-header">
<button type="button" class="navbar-toggle collapsed" data-toggle="collapse" data-target="#navbar" aria-expanded="false" aria-controls="navbar">
<span class="sr-only">Toggle navigation</span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
</button>
</div>
<div id="navbar" class="collapse navbar-collapse">
<ul class="nav navbar-nav">
<li><a href="index.html">Overview</a></li>
<li><a href="organization.html">Organization</a></li>
<li><a href="syllabus.html">Syllabus</a></li>
<li class="active"><a href="assignments.html">Assignments</a></li>
<li><a href="software.html">Software</a></li>
</ul>
</div><!--/.nav-collapse -->
</div>
</nav>
<div class="container">
<div class="page-header">
<div style="float: right"><img width="250" src="images/waterloo_logo.png" alt="University of Waterloo logo"/></div>
<h1>Assignments <br/><small>Data-Intensive Distributed Computing (Winter 2019s)</small></h1>
</div>
<div class="subnav">
<ul class="nav nav-pills">
<li><a href="assignment0.html">0</a></li>
<li><a href="assignment1.html">1</a></li>
<li><a href="assignment2.html">2</a></li>
<li><a href="assignment3.html">3</a></li>
<li><a href="assignment4.html">4</a></li>
<li><a href="assignment5.html">5</a></li>
<li><a href="assignment6.html">6</a></li>
<li><a href="assignment7.html">7</a></li>
<li><a href="project.html">Final Project</a></li>
</ul>
</div>
<h3>Assignment 7: Spark Streaming <small>due 2:30pm April 4</small></h3>
<p>In this assignment, you'll be playing with Spark Streaming. Unlike
the previous assignments that involve a substantial amount of
implementation, the goal of this assignment is to give you some
exposure to Spark Streaming without getting into too much detail. In
other words, this assignment is easier and less time-consuming than
previous assignments, by design.</p>
<p>We'll be working with a dataset released by the New York City Taxi
& Limousine Commission that captures over one billion individual
taxi trips over the past several years. This assignment is inspired by
Todd Schneider's very nice blog post
titled <a href="http://toddwschneider.com/posts/analyzing-1-1-billion-nyc-taxi-and-uber-trips-with-a-vengeance/">Analyzing
1.1 Billion NYC Taxi and Uber Trips, with a Vengeance</a> (worth a
read!). You can find the raw
data <a href="http://www.nyc.gov/html/tlc/html/about/trip_record_data.shtml">here</a>.
The dataset is historic, but we're going to treat it as a data stream
and perform simple analyses using Spark Streaming.</p>
<p>Since the dataset is 100s of GB and too large for an assignment,
we're going to work with a small one-day slice, but that should be
sufficient to give you a flavor of what Spark Streaming is like. The
slice of the dataset is stored in
the <a href="https://github.com/lintool/bespin-data"><code>bespin-data</code></a>
repo. Go ahead and grab the data.</p>
<p>For this assignment, since the dataset is small, everything can be
done in the Linux Student CS environment. That is, there is no
Datasci cluster component for this assignment. We'll be marking your
assignment in the Linux Student CS environment, and you can assume
that we will supply the correct path to this dataset.</p>
<p>Here's what the dataset looks like:</p>
<pre>
$ head -1 taxi-data/part-2015-12-01-0001.csv
green,2,2015-12-01 00:01:03,2015-12-01 00:01:29,N,1,-73.937652587890625,40.804546356201172,-73.940483093261719,40.805999755859375,1,.06,2.5,0.5,0.5,0,0,,0.3,3.8,2,1
</pre>
<p>There are two types of taxi ride records, Yellow and Green.
<p>The schema for yellow taxi rides is as follows:</p>
<pre>
type,VendorID,tpep_pickup_datetime,tpep_dropoff_datetime,passenger_count,trip_distance,pickup_longitude,pickup_latitude,RatecodeID,store_and_fwd_flag,dropoff_longitude,dropoff_latitude,payment_type,fare_amount,extra,mta_tax,tip_amount,tolls_amount,improvement_surcharge,total_amount
</pre>
<p>The schema for green taxi rides is as follows:</p>
<pre>
type,VendorID,lpep_pickup_datetime,Lpep_dropoff_datetime,Store_and_fwd_flag,RateCodeID,Pickup_longitude,Pickup_latitude,Dropoff_longitude,Dropoff_latitude,Passenger_count,Trip_distance,Fare_amount,Extra,MTA_tax,Tip_amount,Tolls_amount,Ehail_fee,improvement_surcharge,Total_amount,Payment_type,Trip_type
</pre>
<p>Each part file contains one minute worth of trips, so for the
entire day there are 1440 part files. Each one of these becomes a part
of a discretized stream to Spark streaming.</p>
<p>Let's start with a very simple query that aggregates the number of
trips by hour. We've implemented this for you in Bespin
in <a href="https://github.com/lintool/bespin/blob/master/src/main/scala/io/bespin/scala/spark/streaming/EventCount.scala"><code>EventCount</code></a>.
Here's how you run it:</p>
<pre>
spark-submit --class io.bespin.scala.spark.streaming.EventCount \
target/bespin-1.0.5-SNAPSHOT-fatjar.jar --input taxi-data --checkpoint checkpoint --output output
</pre>
<p>Before running the Spark streaming job, run the following first, because otherwise you may get "CPU time limit exceeded" errors:</p>
<pre>
ulimit -t unlimited
</pre>
<p>Note that there's a recent bug that's cropped up in OpenJDK and Maven, see <a href="https://stackoverflow.com/questions/53010200/maven-surefire-could-not-find-forkedbooter-class">here</a> and <a href="https://issues.apache.org/jira/browse/SUREFIRE-1588">here</a>. The tl;dr is to skip running test cases when building Bespin, as follows:</p>
<pre>
mvn clean package -Dmaven.test.skip=true
</pre>
<p>The Spark Streaming query itself is simple:</p>
<pre>
val wc = stream.map(_.split(","))
.map(tuple => ("all", 1))
.reduceByKeyAndWindow(
(x: Int, y: Int) => x + y, (x: Int, y: Int) => x - y, Minutes(60), Minutes(60))
.persist()
</pre>
<p>Your first task is to understand exactly what's going on the above
snippet of code: consult
the <a href="https://spark.apache.org/streaming/">Spark Streaming
documentation</a>, or alternatively, there are plenty of resources
online — a simple search will turn up lots of articles and
guides.</p>
<p>Beyond the query itself, the implementation goes through a bunch of
contortions to be able to mock a stream using data from files. There
are a whole host of details, including how to mock a clock so we can
simulate the passage of time. The implementation is based
on <a href="http://blog.ippon.tech/testing-strategy-for-spark-streaming/">this
blog post</a> if you are interested in the details.</p>
<p>The output of each aggregation window is stored in a directory
named <code>output-XXXXX</code> where <code>XXXXX</code> is the
timestamp. The following command will gather all the results
together:</p>
<pre>
$ find output-* -name "part*" | xargs grep 'all' | sed -E 's/^output-([0-9]+)\/part-[0-9]+/\1/' | sort -n
3600000:(all,7396)
7200000:(all,5780)
10800000:(all,3605)
14400000:(all,2426)
18000000:(all,2505)
21600000:(all,3858)
25200000:(all,10258)
28800000:(all,19007)
32400000:(all,23799)
36000000:(all,24003)
39600000:(all,21179)
43200000:(all,20219)
46800000:(all,20522)
50400000:(all,20556)
54000000:(all,21712)
57600000:(all,22016)
61200000:(all,18034)
64800000:(all,19719)
68400000:(all,25563)
72000000:(all,28178)
75600000:(all,27449)
79200000:(all,27072)
82800000:(all,24078)
86400000:(all,18806)
</pre>
<p>There are 24 aggregation windows above, one for each hour.</p>
<p><b>Problem 1</b>: Copy <code>EventCount</code> from Bespin into
your assignment repo under the
package <code>ca.uwaterloo.cs451.a7</code>. You should now be able to
run the following and obtain exactly the same results as above:</p>
<pre>
spark-submit --class ca.uwaterloo.cs451.a7.EventCount \
target/assignments-1.0.jar --input taxi-data --checkpoint checkpoint --output output
</pre>
<p><b>Problem 2</b>: Create a query
called <code>RegionEventCount</code> that counts the number of taxi
trips each hour that drop off at either the Goldman Sachs headquarters
or the Citigroup headquarters. See the
<a href="http://toddwschneider.com/posts/analyzing-1-1-billion-nyc-taxi-and-uber-trips-with-a-vengeance/">Todd
Schneider blog post</a> for context: you're replicating one a
simplified version of one of his analyses. Use these coordinates for
the bounding box of interest:</p>
<pre>
goldman = [[-74.0141012, 40.7152191], [-74.013777, 40.7152275], [-74.0141027, 40.7138745], [-74.0144185, 40.7140753]]
citigroup = [[-74.011869, 40.7217236], [-74.009867, 40.721493], [-74.010140,40.720053], [-74.012083, 40.720267]]
</pre>
<p>To be more precise, you are filtering for taxi trips whose drop off
locations are located within this bounding box and aggregating by
hour. This means that we should be able to run the following job:</p>
<pre>
spark-submit --class ca.uwaterloo.cs451.a7.RegionEventCount \
target/assignments-1.0.jar --input taxi-data --checkpoint checkpoint --output output
</pre>
<p>And the following command should gather the answers:</p>
<pre>
$ find output-* -name "part*" | xargs grep 'goldman' | sed -E 's/^output-([0-9]+)\/part-[0-9]+/\1/' | sort -n
21600000:(goldman,?)
25200000:(goldman,?)
28800000:(goldman,?)
...
$ find output-* -name "part*" | xargs grep 'citigroup' | sed -E 's/^output-([0-9]+)\/part-[0-9]+/\1/' | sort -n
3600000:(citigroup,?)
7200000:(citigroup,?)
10800000:(citigroup,?)
...
</pre>
<p>Use exactly "goldman" and "citigroup" for the names of the
keys. The actual counts should appear in place of the "?" above.</p>
<p>As a hint, <code>RegionEventCount</code> requires minimal
modifications over <code>EventCount</code>; basically, add a filter,
and that's it.</p>
<p><b>Problem 3</b>: Let's build a simple "trend detector" to find out
when there are lots of arrivals at either Goldman Sachs or Citigroup
headquarters, defined in terms of the bounding boxes, exactly as
above. We'll consider intervals of ten minutes, i.e., 6:00 to 6:10,
6:10 to 6:20, etc. The trend detector should "go off" when there are
at least twice as many arrivals in the current interval as there are
in the past interval. To reduce "spurious" detections, we want to make
sure the detector only "trips" if there are <i>ten or more</i>
arrivals in the current interval. That is, if there are two arrivals
in the last ten minute interval and four arrivals in the current ten
minute interval, that's not particularly interesting (although the
number of arrival has indeed doubled), so we want to suppress such
results.</p>
<p>Call this program <code>TrendingArrivals</code>. We'll run it as
follows:</p>
<pre>
spark-submit --class ca.uwaterloo.cs451.a7.TrendingArrivals \
target/assignments-1.0.jar --input taxi-data --checkpoint checkpoint --output output &> output.log
</pre>
<p>For simplicity, the detector should output its results
to <code>stdout</code> in the form of the following:</p>
<pre>
Number of arrivals to Goldman Sachs has doubled from X to Y at Z!
</pre>
<p>or</p>
<pre>
Number of arrivals to Citigroup has doubled from X to Y at Z!
</pre>
<p>Where <code>X</code> and <code>Y</code> are the number of arrivals
in the previous and current interval, and <code>Z</code> is the
timestamp of the current interval. These timestamps are exactly the
same form as above, e.g., <code>21600000</code>.</p>
<p>In other words, when we're marking, we'll be
grepping <code>output.log</code> for the phrase "Number of arrivals to
Goldman Sachs" and "Number of arrivals to Citigroup", so please make
sure that the output is in the format we expect.</p>
<p>Also, the program should output the status for each batch to the directory specified by the <code>output</code> argument. Each status is stored in a separate file with the name of the format <code>part-${timestamp}</code> where timestamp is a 8-digit string padded with leading zeros.
The following command should gather the answers:
<pre>
$ cat output/part-*/* | grep "(citigroup"
(citigroup,(${Current value},${Current timestamp},${Previous value}))
...
$ cat output/part-*/* | grep "(goldman"
(goldman,(${Current value},${Current timestamp},${Previous value}))
...
</pre>
<p>As a hint, <code>TrendingArrivals</code> is a simple modification
to <code>RegionEventCount</code>. You'll need to change the window
definition (from hourly to 10 minutes) and learn how to
use <code>mapWithState</code>, but that's basically it.</p>
<h4 style="padding-top: 10px">Turning in the Assignment</h4>
<p>Please follow these instructions carefully!</p>
<p>Your implementations should go in
package <code>ca.uwaterloo.cs451.a7</code>. Make sure your
implementation runs in the Linux Student CS environment. The following
check script is provided for you:</p>
<ul>
<li><a href="assignments/check_assignment7_public.py"><code>check_assignment7_public.py</code></a></li>
</ul>
<p>When you've done everything, commit to your repo and remember to
push back to origin. You should be able to see your edits in the web
interface. Before you consider the assignment "complete", we would
recommend that you verify everything above works by performing a clean
clone of your repo and run the public check scripts.</p>
<p>That's it!</p>
<h4 style="padding-top: 10px">Grading</h4>
<p>The entire assignment is worth 30 points:</p>
<ul>
<li>Problem 1 is worth 5 points.</li>
<li>Problem 2 is worth 10 points.</li>
<li>Problem 3 is worth 15 points.</li>
</ul>
<p style="padding-top: 20px"><a href="#">Back to top</a></p>
<div style="padding-bottom: 100px"></div>
</div><!-- /.container -->
<!-- Placed at the end of the document so the pages load faster -->
<script src="https://ajax.googleapis.com/ajax/libs/jquery/1.12.4/jquery.min.js"></script>
<script src="js/bootstrap.min.js"></script>
<!-- IE10 viewport hack for Surface/desktop Windows 8 bug -->
<script src="js/ie10-viewport-bug-workaround.js"></script>
</body>
</html>