An RDMA Middleware for Multi-stage
Asynchronous Shuffling in Analytical
Processing
Rui C. Gonçalves1, José Pereira1, Ricardo Jiménez-Peris2
1INESC TEC & U. Minho

2Univ. Politécnica de Madrid & LeanXcale

Distributed Applications and Interoperable Systems
June 2016

Context› Big Data: Explosion of data being generated
and stored every da› Holds valuable knowledge for organizations'operatio› Opportunity to improve their e cienc› New challenges to store and process massive
amounts of datayffiyn2

Context› Emergence of new solutions for large scaledata processing, as alternatives to traditional
Relational Database Management System› NoSQL database system› MapReduce programming mode
› Stream processing framework› New solutions that o er SQL-like interface› Hiv› Impalasslss

ffe

3

Shuffling› Shu

ing is a key concept in multiple solution› Provides distribution of data for parallel processing,possibly aggregating related item

› Di erent shu

ing approache› Push-based vs pull-based› Synchronous vs asynchronou› Hash code vs rando› Requires distributed coordinatio› Critical for systems performancesnsss

m

ffl

ffl

ff

4

Shuffling› Shu

ing is a key concept in multiple solution› Provides distribution of data for parallel processing,possibly aggregating related item

› Di erent shu

ing approache› Push-based vs pull-based› Synchronous vs asynchronou› Hash code vs rando› Requires distributed coordinatio
Leveragefrom
RDMA
Criticalforsystems
performance›
protocols to improve shuf ing
sflnsssm

ffl

ffl

ff

4

RDMA Protocols› Low latency, high throughput
communication› Direct application access to network
hardwar› Zero-copy communication› Multiple RDMA technologie› Software implementations also available: e.g.
Soft-iWARP and Soft-RoCEssse5

RDMA Verbs› Asynchronous API to use RDMA protocol› Requests are queued, and completion events may
be generated when requests are completed› Two types of communication semantics› Memory semantics (one-sided read/writeoperations› Channel semantics (two-sided send/receive
operations› Prede ned memory location› Requires redesigning applications currently usings:s))

socket-based communications
fi6

Shuffling for Analytical Processing

7

Shuffling for Analytical Processing› Shu

ing implementation tosupport analytical workloads on aJava distributed query engine (SQL)› Parallelization of stateful operators
requires shu ing to group related rows

ffl

ffl

7

Shuffling for Analytical Processing› Shu

ing implementation toselect *from a inner join b
on a.x = b.x

support analytical workloads on aJava distributed query engine (SQL)

Shuffle› Parallelization of stateful operators

HashJoinrequires shu ing to group related rows

Hashmatchingkeys

ffl

ffl

7

ShuffleShuffleLocalScanLocal
Scan

Shuffling for Analytical Processing› Shu

select *from a inner join b
on a.x = b.x

ing implementation tosupport analytical workloads on aJava distributed query engine (SQL)

Shuffle› Parallelization of stateful operators

Hash
Join

requires shu ing to group related rows

› Multi-stage, to allow multiple shuing

Hashmatching
keys

steps

› Asynchronous, to allow tasks to
execute in parallel at di erent steps

ffl

ff

ffl

ffl

7

ShuffleShuffleLocal
ScanLocal
Scan

DQE Analytical Processing
Architecture› Workers execute the operators of the query plan,
which requires fetching rows from child operator› When fetching rows from Shu e operators› It may return a row received from other worke› If there is no received row, it fetches a row from its
child operato› The row fetched may be returned or shu
DQE Instance

DQE Instance

DQE InstanceWorkerWorkerWorkerWorker...

Local
Data

WorkerLocal
Data...

Worker...

ed

Local
Data

...

s:r

fflr

8

ffl

NetworkAsynchronous Shuffling› Push-based asynchronous shuin› Rows are sent as soon as they are processed bythe Shu e operato› Workers maintain Shu

e Queues, which receive

remote rows and store them until they are
consumed
DQE Instance

DQE Instance

DQE InstanceWorkerWorkerWorkerShuffle Queues

Shuffle Queues

Shuffle Queues

Local
Data

WorkerShuffle Queues

Worker

Local
Data

...

WorkerShuffle Queues...

Shuffle Queues

...

...

gffl

r

ffl

9ffl

Network

Local
Data

Shuffle Queues› Contain an incoming and an outgoing circularbu er per each remote threa› Outgoing bu ers are used to serialize rows to sen› Incoming bu ers are used to receive rows and store it
until they are consume› Rows from local workers are put on a dynamic queu

› Communication middleware transfers data froman outgoing bu er to the matching incoming
DQE Instancebu er
WorkerLocal
Data

in.
out.
in.

Shuffle Queues

out.
e...
d

ff

ff

ff

ff

ff

10

d...

d

Worker...

local

Shuffle Queues

Communication Middleware› Key functionalities require› Ability to send (and queue) rows to remoteworker› Ability to retrieve queued row› Ability to block a worker when there are no rows
to process (and to wake it up when new rows
become available› Ability to block a worker when the local outgoingbu ers are full (and to wake it up when spacebecomes available)ds)sff

11

RDMA Middleware
Overview› RDMA Write operations to transfer dat› Send/Receive operations for noti cation› Dedicated network thread on each DQE
instanc› Tracks requests' completion event› Manages noti cations

safisfie12

Workers› When rows to shu

e become available› Serialize rows and post write requests, unless› Outgoing bu er is full (worker blocks› Remote incoming bu er is full (worker proceedswith its operation› When polling rows from shue queues› Check for rows received in local queu
› If no row was available, poll incoming bu er› When data is removed from bu ers, noti es the::d:sffefi)

ffl

ff

fflff)

ff

ffl

13

remote side, so that space released can be reuse› Block if shu e queues are empty and child operator
has no more rows

Network Thread› In charge of operations that follow thecompletion event of network operation› After completion of write operation› Noti es the remote side that data was writte› Releases space on outgoing bu er› Wakes up workers blocked for writingnsssfffi14

Network Thread› After receiving noti cation› Maintains a queue of bu ers with data available
based on write noti cations (to avoid the needof active polling› Wakes up workers blocked for readin› Maintains the space available on remote bu ers
based on read noti cations (to determine
whether write requests can be performed› Post write requests for data on bu ers (if worker was
unable to post a write request previously)ff)g

ffsfffififi)

15

Implementation Decisions› Single connection between each pair ofmachine› Multiple workers share the same connection/requests queu

› RDMA write (vs send/receive› Avoids the need of a rendezvous protoco› Less CPU usage (remote side)l)es

16

Optimizations› Batch multiple rows/noti cations before
transferring dat› Trades latency for reduced contention accessing
network resource› Pre-initialized data structure› Data structures initialized during creation of JDBC
connection (reused for multiple queries› Reduce overheads of JNI calls by using statefulverbs methods provided by jVerbs library)

sfias

17

Performance Evaluation
Synthetic Query Plan
40

n := n * 3/2

Shufflen := n * 3/2

ShuffleSockets
RDMA

Shuffle32

Execution5 time5(s)Shuffle

38.66

25.46
22.15

24

23.54

22.05

16
7.42

8

5.97

5.81

5.65

5.91

0Int
Generator

Int
Generator

816

32

48

64

#5workers› 8 machines (Intel Core i3 CPU, 2 cores, 8GB RAM, 1 GigE
› Query plan where each machine shu es 5M integers twic
› RDMA middleware compared with a socket middlewar› RDMA middleware: execution time ~3.9x fastere)effl18Performance Evaluation
SQL Join Query
200
180

Sockets

170.84170.84

RDMA160

Execution5 time5(s)select ol_o_id, ol_w_id, ol_d_id,
sum(ol_amount) as revenue,
o_entry_dfrom order_line, orders, new_order, customer
where o_entry_d > timestamp('2013-07-01 0.00.00')
and c_id = o_c_id and c_w_id = o_w_id
and c_d_id = o_d_id and no_w_id = o_w_id
and no_d_id = o_d_id and no_o_id = o_id
and ol_w_id = o_w_id and ol_d_id = o_d_id
and ol_o_id = o_id and c_state like 'A%'
group by ol_o_id, ol_w_id, ol_d_id, o_entry_dhaving revenue > 80000.00
order by revenue desc, o_entry_d;

140
120
100
80
46.54 42.69

60
40

22.76 20.19

17.17 15.42

16.52 14.45

17.22 14.57

1632

64

96

2000
18

#5workers› Triple hash join with aggregation quer
› Uses 8 shu e operator› RDMA middleware: execution time ~1.14x faster
ys

ffl

19

Performance Evaluation
SQL Aggregation Query
200160Execution5 time5(s)select ol_number, sum(ol_quantity) as sum_qty,sum(ol_amount) as sum_amount,sum(ol_quantity) / count(*) as avg_qty,sum(ol_amount) / count(*) as avg_amount,count(*) as count_orderfrom order_line
where ol_delivery_d > timestamp('2013-07-01 0:00:00')
group by ol_number
order by sum(ol_quantity) desc;

180

Sockets
159.67159.67

RDMA

140
120
100
80
60

33.11 31.19

40
20

16.35 14.68

12.39 11.26

11.81 11.13

11.89 11.28

1632649600
18#5workers› Aggregation quer
› Uses 2 shu e operator› RDMA middleware: execution time around 1.06x fastersyffl

20Summary› Leverage from RDMA to improve performanceof shu ing in large scale analytical processin› Design an RDMA communication middleware tosupport push-based asynchronous shu ing› Improves time spent on communication
operations by 3.9x, when compared with asockets-based middlewar› Using an RDMA approach is bene cial evenwhen using software RDMA implementationsgfflfieffl21