SOLUTIONS AND MARKING SCHEME
January 2019-2020
ECS765P Big Data Processing Duration: 2 hours
YOU ARE NOT PERMITTED TO READ THE CONTENTS OF THIS QUESTION PAPER UN-
TIL INSTRUCTED TO DO SO BY AN INVIGILATOR.
Instructions:
This paper contains FOUR questions. Answer ALL questions
Calculators are permitted in this examination. Please state on your answer book the
name and type of machine used.
COMPLETE ALL ROUGH WORKINGS IN THE ANSWER BOOK AND CROSS THROUGH
ANY WORK WHICH IS NOT TO BE ASSESSED.
IMPORTANT NOTE:
THE ACADEMIC REGULATIONS STATE THAT POSSESSION OF UNAUTHORISED MA-
TERIAL AT ANY TIME WHEN A STUDENT IS UNDER EXAMINATION CONDITIONS IS AN
ASSESSMENT OFFENCE AND CAN LEAD TO EXPULSION FROM QMUL.
PLEASE CHECK NOW TO ENSURE YOU DO NOT HAVE ANY NOTES, MOBILE PHONES
OR UNATHORISED ELECTRONIC DEVICES ON YOUR PERSON. IF YOU HAVE ANY
THEN PLEASE RAISE YOUR HAND AND GIVE THEM TO AN INVIGILATOR IMMEDIATELY.
PLEASE BE AWARE THAT IF YOU ARE FOUND TO HAVE HIDDEN UNAUTHORISED MA-
TERIAL ELSEWHERE, INCLUDING TOILETS AND CLOAKROOMS IT WILL BE TREATED
AS BEING FOUND IN YOUR POSSESSION. UNAUTHORISED MATERIAL FOUND ON
YOUR MOBILE PHONE OR OTHER ELECTRONIC DEVICE WILL BE CONSIDERED THE
SAME AS BEING IN POSSESSION OF PAPER NOTES. MOBILE PHONES CAUSING A
DISRUPTION IS ALSO AN ASSESSMENT OFFENCE.
EXAM PAPERS MUST NOT BE REMOVED FROM THE EXAM ROOM.
Examiners:
Dr Felix Cuadrado and Dr Johan Pauwels
© Queen Mary University of London, 2019-2020
Page 2 ECS765P (2019-2020)
Question 1
You have a dataset from a non-profit organisation listing the full history of member contri-
butions for emergency appeals when e.g. a disaster happens.
For the next emergency appeals campaign, the organization is aiming to send postal
envelopes to the 500 highest contributors up to this point.
The dataset is a collection of rows, each one registering an individual contribution from a
member in the following format:
timestamp;memberId;isMultipleDonor;donationAmount;memberAge
The isMultipleDonor field is a boolean field that will be True if the member has made
more than one donation. donationAmount contains the value of that individual donation,
whereas memberAge includes for how many months they have been a member.
(a) (i) Design a combination of *two* MapReduce programs that computes the top 500
members with highest individual average contributions to these urgent appeal
campaigns. You should only consider members with more than one individual
contribution.
The code flow must be explained, discussing the input and output of each Map
and Reduce function that has been defined (without detailing the implementation).
You may use a diagram to illustrate the overall data flow.
Note: while a single MapReduce solution is technically possible, it would result
in potentially some scalability/performance issues. Single MapReduce solutions
should mention these aspects specifically.
(ii) Write the pseudocode for one of the two MapReduce jobs you designed in the
previous question. State in your solutions any assumptions that are made as
part of the program, as well as the behaviour of any custom function you deem
necessary.
[15 marks — basic]
Solution:
The first program will compute the average contribution of each member from
the dataset that is a multiple times donor 3 . That input will be fed to the second
program that will compute the global average (numerical summarisation) of all
these results 3 . First mapper will emit (memberId, donationAmount) 1 , First
reducer will emit (memberId, averageAmount) 1 Second mapper will emit (None,
(memberId,averageAmount)), 1 Second reducer will emit 500 x (memberId,
ranking, averageAmount) 1
Either of these two MR programs is sufficient for the 5 marks
Turn over
ECS765P (2019-2020) Page 3
def mapper1(self, _, line):\mk{1}
fields = line.split(";")
multipleDonor = fields[2]
if(multipleDonor):\mk{1}
memberId = fields[1]
amount = fields[3]
yield (memberId, amount)\mk{1}
def reducer1(self, memberId, values):\mk{1}
average = computeAverage(values)
yield(memberId, average) \mk{1}
def mapper2(self, memberId, average): \mk{1}
yield(None, (memberId, average)) \mk{1}
def reducer2(self, _, values): \mk{1}
sorted = values.sortByValue()
top500 = sorted.getTop(500)\mk{1}
for item in top500:
yield(item[0], item[1]) \mk{1}
(b) This question is about the Combiner in the context of MapReduce jobs.
(i) Explain who runs the Combiner function and at what point in time Combiner
functions execute during the shuffle and sort stage of MapReduce jobs.
(ii) Discuss whether the *second* job from 1a) would benefit from using a Combiner.
To do so, first list what will change when a Combiner runs, and then explain its
performance impact. If you completed 1a) with a single MapReduce job, then
answer the question referring to that job.
[10 marks — medium]
Solution:
A Combiner is an intermediate Reducer that is executed at each of the Mapper
nodes 2 . The Combiner runs after the Mappers have finished their execution,
and before data is transferred to the Reducers 2 . Combiner runs according to
Hadoop from 0 to multiple times, as it’s optional and idempotent 1 .
The second job can run the same Reducer as a Combiner 2 . This will prevent
memory issues when running the job 1 , and make it complete considerably
faster, as a single Reducer invocation would need to sort every single item that
Turn over
Page 4 ECS765P (2019-2020)
needs to be ranked. 2
Turn over
ECS765P (2019-2020) Page 5
Question 2
(a) You have a dataset of user ’reactions’ from the Instagram social media network. The
dataset contains one entry for each time a user liked a specific post, recording the
following information:
timestamp;userId;storyId;reactionType;
reactionType records the type of emoji users reacted to the message, with potential
types being ”like”, ”dislike”, or ”love”
This dataset is given as input to the following MapReduce job:
def mapper(self, _, line):
fields = line.split(";")
id = fields[1]
reactionType = fields[3]
yield (id, (reactionType,1))
def reducer(self, id, reactions):
reactionCounts = sumReactionsByType()
#sumReactionsByType() sums the partial totals
# for each Type, obtaining a dictionary of
# types and counts for that reaction
top = getMax(reactionCounts)
yield ( id, top.getType())
What will be the outcome of the MapReduce job? Explain the outcome as well as the
nature of information exchanged in the job execution.
[5 marks — basic]
Solution:
The outcome will be the most common type of reaction for each post that has
been posted in the dataset 3 . The mapper sends for each reaction the post id
as a key, and a pair with the type of reaction and one as a pair 1 . The reducer
receives all the reactions for a post, aggregates them together by type, and yields
the most common one 1 .
(b) Suppose the program presented in 2a) will be executed on a dataset of 5 billion
(5,000,000,000) reactions, collected over 3*365 days of data. In total there are
50,000,000 stories that have been reacted to. The total size of the dataset is 4*128
Gigabytes. The cluster has 5000 worker nodes (all of them idle), and HDFS is
configured with a block size of 128MB.
Using that information, provide a reasoned answer to the following questions. State
Turn over
Page 6 ECS765P (2019-2020)
any assumptions you feel necessary when presenting your answer.
Note: you can consider in your computations either 1kB = 1000 bytes, or 1 kB = 1024
bytes.
(i) How many physical nodes will be involved during the execution of the Map and
Reduce tasks of the job? How many input splits are processed at each physical
node?
(ii) In each single worker node, how many times does the Map method run?
(iii) How many times will the Reduce method be invoked at each reducer node?
[10 marks — medium]
Solution:
There is one data split per HDFS block: 4*128GB/128MB = 4000 splits that need
to be processed. The ApplicationMaster will attempt to assign one to a node,
taking 4000 nodes for Map (ideally local nodes with the data already stored) 2 .
As there is enough capacity, one split will be processed by each involved physical
node 2 . The number of reduce nodes is user defined, by default 2 1
If we assume balanced size of all splits in number of lines, there will be 5 billion
lines/4000 mappers = 1.25 million lines per Mapper 3 .
Students can either use nreducers, or give it a value. The number of unique
keys in this case is the 50 million stories. Number of reduce invocations is 50
million/nred (25m with the default 2) 2 .
(c) Define the concept of data skew in distributed data processing. For the program
defined in 2a) and the data and environment figures presented in 2b), discuss whether
data skew would cause a noticeable effect while running the provided code. If that
will be the case, mention (no need to fully detail) one potential change to the job that
would greatly improve the skew effect.
[10 marks — advanced]
Solution: Data skew affects MapReduce jobs by having an unbalanced distri-
bution of values between the Reduce keys, hence unbalancing the distributed
load. 3 In this particular case, the dataset being of social interactions, data skew
will be a notable effect, with some posts being vastly more popular than others.
This will result in potentially heavy data skew. 3 A simple way to palliate this skew
would be to add a Combiner to the mappers that aggregates together the counts
of reactiontypes for the same post id. 2 As long as posts are emitted by the same
mapper, this approach will be effective. 2
Turn over
ECS765P (2019-2020) Page 7
Question 3
(a) (i) Give one example of a Spark element-wise operation. Describe how would the
Reduce step of an equivalent Hadoop job look like.
(ii) Based on your previous answer, discuss the parallelisation potential of this
element-wise operation as a function of the size of the input data size. De-
tail in your explanation what nodes will be active during the computation, as well
as the data transferred over the network.
[9 marks]
Solution: i) Example: map, flatMap, filter. The Reduce step is not needed, i.e
element-wise operations are Map-only jobs 3 . ii) An element-wise or map-only
operation is embarrassingly parallel, which means that scaling up to a larger
cluster is trivial 2 . All nodes that contain a partition of the data will be used in the
computation 2 . Since no shuffling of the data is needed, no transfers over the
network are needed 2 .
Marking scheme: i) Three marks: basic i)) Six marks: basic-medium
(b) You are given a dataset with the historical results of the top-tier English Football
League per season. It is in a CSV (comma separated value) format and consists of
the following fields:
Season ,Team, Wins , Draws , Losses , GoalsFor , GoalsAgainst
An example of this data is as follows:
1888−1889, Preston North End,18 ,4 ,0 ,74 ,15
1888−1889,Aston V i l l a ,12 ,5 ,5 ,61 ,41
1888−1889,Wolverhampton Wanderers ,12 ,4 ,6 ,51 ,37
1888−1889, Blackburn Rovers ,10 ,6 ,6 ,66 ,45
1888−1889, Bol ton Wanderers ,10 ,2 ,10 ,63 ,59
Suppose you execute the following Python Spark job on the dataset:
1l i n e s = sc . t e x t F i l e ( ’ hdfs : / / inputPath . csv ’ )
2f i l t e r e d 1 = l i n e s . f i l t e r ( lambda l : len ( l . s p l i t ( ” , ” ) ) == 7)
3rows = f i l t e r e d 1 .map( lambda l i n e : l i n e . s p l i t ( ” , ” ) )
4f i l t e r e d 2 = rows . f i l t e r ( lambda r : r [ 2 ] == r [ 4 ] )
5mapped = f i l t e r e d 2 .map( lambda r : ( r [ 1 ] , 1 ) )
6r e s u l t = mapped . reduceByKey ( lambda a , b : a+b )
7r e s u l t . saveAsTextFi le ( ’ hdfs : / / outputPath ’ )
We provide for reference the following information about the Python and Spark-specific
functions appearing in the program:
Turn over
Page 8 ECS765P (2019-2020)
filter is a Spark transformation that for each input element will either generate the
same element as output if the condition expressed in the function is true, or no element
at all if it is false.
map is a Spark transformation that for each input element generates one output
element, whose value is dictated by the function.
reduceByKey is a Spark transformation that takes as input an RDD of pairs of
elements, first groups all the pairs, putting together the ones with the same key, and
generates one pair as a result of the group. The key of the result is the common key,
and the value is the result of reducing the group of values into a single one by applying
the function.
min is a Spark action that returns to the driver the minimum item as specified by the
optional key function.
saveAsTextFile is a Spark action that saves the RDD to the provided HDFS path.
len is the Python function that returns the length of a string, or list.
split is the Python function that splits a string into a list of strings, using the provided
argument as separator.
int is the Python function that converts the string representation of a number into its
numerical value.
(i) Explain what the final result of the code will be. Discuss what each line in the
code does to end up at this result. It may be easier to reference each of these
lines by their number or variable names.
(ii) Detail the changes you need to make to the above code in order to find what
season had the least number of draws in history. The output needs to be a string
of the form “The least number of draws (N) happened in season YYYY-YYYY”. You
may explain these changes in your own words, or provide the code in pseudocode
or Python.
[16 marks]
Solution:
i) The code returns the number of seasons each team had as many wins as
losses 2 . Line 1 creates an RDD by reading a text file from HDFS 1 . Line 2
filters the data for lines which do not have the correct format/have missing fields 1 .
Line 3 splits CSV lines into separate values 1 . Line 4 keeps only the seasons
when a team had as many wins as losses 1 . Line 5 creates key-value pairs of
the team name and a value of 1 of the output of the previous line 1 . Line 6 sums
the values of 1 per team name to count the number of times each time has had
as many wins as losses in a season 1 . Line 7 is an action that writes the result
to HDFS 1 .
ii) Something along the lines of:
Turn over
ECS765P (2019-2020) Page 9
1l i n e s = sc . t e x t F i l e ( ’ hdfs : / / inputPath . csv ’ )
2wel l formed = l i n e s . f i l t e r ( lambda l : len ( l . s p l i t ( ” , ” ) ) == 7)
3rows = wel l formed .map( lambda l : l . s p l i t ( ” , ” ) )
4season draws = rows .map( lambda r : ( r [ 0 ] , i n t ( r [ 3 ] ) ) )
5draws per season = season draws . reduceByKey ( lambda x , y : x+y )
6r e s u l t = draws per season . min ( key=lambda x : x [ 1 ] )
7answer = ’ The l e a s t number o f draws ({} ) happened i n
8season {} ’ . format ( r e s u l t [ 1 ] , r e s u l t [ 0 ] )
Marking scheme: i) Nine marks: basic-medium. ii) Seven marks: medium-
advanced. Line 4 2 , line 5 2 , line 6 2 , line 7–8 1 . Similar code will be given
equivalent marks as long as they implement analogous functions.
Turn over
Page 10 ECS765P (2019-2020)
Question 4
(a) This question is about different types of stream processing.
(i) Components of a streaming algorithm are either stateless or stateful. Explain
what these terms mean in this context and give an example computation for each.
(ii) Micro-batching and pure stream processing both run on unbounded data. Explain
how these two processing models differ. Discuss in what situation each would be
more appropriate, by making reference to latency and throughput considerations.
[12 marks — medium]
Solution:
i) A stateless computation is one where the function will be run on each tuple 1
without the influence or ’state’ maintained from any other 1 , an example of this
would be a filter 1 . Stateful operations may maintain a state 1 based on tuples
previously seen 1 , for example an aggregate count for keys 1 .
ii) Analysis ran in a pure stream processing fashion will operate on every tuple
as they arrive 1 , leading to low latency of results 1 . Micro-batching on the
other-hand may wait x number of seconds before triggering 1 the operation and
executing on all ingested tuples, often leading to higher throughput, but obviously
slower return of the result 1 . Pure streaming may be more appropriate when
there is some real-time requirement or alerts 1 , whilst micro-batching may be
more appropriate for windowing 1 .
(b) This question is about the scalability of streaming systems.
(i) Streaming systems often consist of data sources and computational units; Spouts
and Bolts respectively in Apache Storm. Describe two things that may happen to
these components when a system is scaled up from a single node to a cluster of
machines for higher throughput.
[6 marks — basic]
Solution:
i) The first thing that may happen to the bolts and spouts is that they are distributed
among the nodes within the cluster instead of being on the same machine 3 . The
second thing to occur would be replication of the bolts to allow their tasks to be
parallelised 3 .
(c) Graphs contain a set of vertices (entities within the data) alongside a set of edges
(relationships between entities). Parallel Breadth First Search is a popular algorithm
for resolving the shortest path between two entities within a graph.
Explain how the Pregel model (think like a vertex) can be used to implement graph
algorithms like Parallel Breadth First Search in a distributed environment such as
Turn over
ECS765P (2019-2020) Page 11
Spark GraphX. Give one example use case of finding the shortest path between
entities.
[7 marks — advanced]
Solution:
i) In the Pregel model computation is carried out in supersteps 1 . Each vertex
gets a function to run every superstep, within this they may ingest messages,
calculate some internal state, send message’s to neighbours and vote to halt 2 .
Each vertex may execute in parallel and may, therefore, be distributed across any
number of machines 2 . An example algorithm for shortest path is routing traffic
between two nodes in a computer network 2 .
End of questions

欢迎咨询51作业君