日本ソフトウェア科学会第 33 回大会 (2016 年度) 講演論文集

Describing Pregel Algorithms with
Non-adjacent Data Communication
Yongzhe Zhang

Hsiang-Shang Ko

Zhenjiang Hu

Pregel is a popular parallel computing model for dealing with large-scale graphs. However, it can be tricky
to implement traditional graph algorithms correctly and eﬃciently in Pregel’s vertex-centric model, as the
programmer needs to carefully restructure an algorithm in terms of supersteps and message passing, which
are low-level and detached from the algorithm description. There exist domain-specific languages that hidemessage passing behind the scenes, and let the programmer express, at a higher level, their intention offetching data from adjacent vertices. But only allowing communication between neighbors is insuﬃcient
to express a still wide range of graph algorithms. In this paper, we design a new domain-specific language
for Pregel that allows the programmer to fetch or modify data on non-adjacent vertices. We give several examples to show that this feature allows many graph algorithms to be expressed more naturally andconcisely.

equipped with a failure recovery mechanism, which

1 Introduction

is important for programs running in a distributed

The rapid increase of graph data in real world

environment.

calls for eﬃcient analysis on massive graphs. How-

Although Pregel is well-suited for parallelism andever, graph computation is in general diﬃcult tohave good scalability, implementing graph algo-parallelize or scale, due to the inherent interde-rithms in Pregel is challenging. As summarized

pendencies in graph data.

Google’s Pregel [11]in several papers [3] [6], the following reasons makeis a framework proposed to process such large

programming in Pregel inconvenient to program-

graphs, which runs on shared nothing architec-

mers:

ture and implements the bulk-synchronous parallel• Explicit state control: In Pregel, programmers(BSP) model [14] to handle large-scale graphs using

write a function compute() that describes thethe vertex-centric computing paradigm. In Pregel,

local computation performed on the vertices.

computation on graphs is split into supersteps, in

Usually that function needs to refer to a global

each of which the vertices run in parallel perform-

state to decide what part of the computation

ing some local computation, and data communi-

should be performed, and global communica-

cation takes place between supersteps. Pregel istion is needed to perform state transition whennecessary.

This is an unrefereed paper. Copyrights belong to
the Author(s).
Yongzhe Zhang, Zhenjiang Hu, 総合研究大学院大学複合科学研究科, School of Multidisciplinary Sciences,
SOKENDAI.
Hsiang-Shang Ko, 国立情報学研究所, National Instituteof Informatics.

• Explicit message passing: The utilization of
message passing between vertices is hard to understand, because it is hard to see where themessage is from and how the message is consumed without searching through the wholecompute() function. In addition, Pregel onlyin graph computing. These two restrictions pre-allows a single message type, which usually

clude a wide range of Pregel algorithms from being

contains diﬀerent contents for diﬀerent commu-

implemented in those DSLs.

nication purposes, and is hard to comprehend
as well.

As an example, let us consider the ShiloachVishkin Practical Pregel Algorithm (S-V PPA) [15],• Manually arranged supersteps:

In Pregel,

which calculates the connected components of an

the

Bulk-

undirected graph in a logarithmic number of su-

Synchronous Parallel model, one conceptual

persteps. This algorithm builds the connectivity in-step in a high-level algorithm description may

formation using the disjoint-set data structure [4].

have to be arranged manually into multiple

Specifically, the data structure is a forest, and thesupersteps, which can be drastically diﬀerent

vertices in the same tree are regarded as belong-from the original conceptual step and obscureing to the same connected component. Each vertex

the logic.

maintains a pointer that either points to some other

which

is

a

specializationof

It is inconvenient and error-prone to let program-vertex in the same connected component, or points

mers keep all those detailed issues in mind when im-

to itself, in which case the vertex is the root of aplementing a graph algorithm in Pregel. Further-tree. S-V PPA is an iterative algorithm that be-

more, having lots of low-level details in the codegins with a forest of n root nodes, and in each step

will greatly hurt the readability of the program,it tries to discover edges connecting diﬀerent trees

and make the program hard to maintain when someand merges the trees together. The description ofchanges need to be made at the algorithm level.

the algorithm is shown in Figure 1.

Several domain-specific languages (DSLs) have

To implement this algorithm in Pregel, one chal-

been proposed to provide more intuitive ways oflenge is to correctly translate the intention of fetch-

implementing graph algorithms in Pregel, such as

ing or modifying non-local data into message pass-

Green-Marl [7] and Fregel [3].

These DSLs all

ing and arrange the supersteps. For example, one

aim to partially solve the problems listed above.

specific intention is for u to compare its parent’s

Specifically, they aim to simplify programming in

ID and pointer to know whether its parent is aPregel by allowing programmers to write the pro-root node. Before doing the comparison, u’s par-gram compositionally to avoid explicit state con-ent should send its pointer to u, but it is u thattrol, hide explicit message passing by data fetching,

maintains the parent-child relationship, so u’s par-generate supersteps according to data dependen-

ent actually does not know to which vertex to sendcies, and automatically produce the message type.

that message. Therefore, u should first send a re-

These two DSLs still have several severe restric-

quest to u’s parent, and then in the next super-tions, however: They only allow data fetching along

step, u’s parent replies to all its children. In thethe edges in a graph, so that only a simple “flip-

third superstep, u can finally do the comparison toping” from data fetching to message sending ischeck whether its parent is a root node or not. We

needed when compiling their programs to Pregel.

see that this intention, despite its simplicity, has

But in general, a step in an algorithm may need

to be translated into three supersteps containing a

to fetch data from arbitrary vertices. Then, there

query-reply conversation between each vertex andis no way for a vertex to modify the state of anits parent, and it is not until the last superstep that

arbitrary vertex, which is also a common pattern

the comparison can be done.

Input: an undirected graph
Output: connected components represented
as disjoint setsin this model, each vertex can fetch local or nonlocal data, perform local computation, and modify
local or non-local state. This notion is more highlevel than Pregel supersteps, in which the vertices

set all vertices as root nodes;

have to explicitly handle the messages, and is more

repeatnatural for describing Pregel algorithms. Then, we

for each vertex u do
if u’s parent is a root node thenchoose a neighbor v whose parent’s
ID is smaller than u’s parent’s ID;
if v exists, let u’s parent point to v’s
parent;else
let u point to its own grandparent;
end
end
until the disjoint-set structure does notdesign a new DSL to allow programmers to express
algorithms in our model, and sketch a transformation from our DSL to Pregel. The concise syntax
of our DSL makes it easy to implement more complicated graph algorithms. The main contributionsof this work are:
• We introduce the concept of algorithmic superstep, which is more natural for describing
Pregel algorithms at a higher level.
• We design a DSL that allows programmers to
describe graph algorithms in terms of algorith-change;mic supersteps, and provide concise syntax for
図1

S-V PPA algorithm description

fetching or modifying non-local state, improving on the expressiveness of the existing DSLs.

Unfortunately, such pattern actually cannot be• We give a representative example to illustrate

implemented in the DSLs mentioned above. The

how to use the DSL and sketch a transforma-reason is that S-V PPA maintains a dynamic inter-tion from our DSL into Pregel.

nal data structure that is orthogonal to the original

The organization of the rest of the paper is as fol-graph and requires communication along this inter-lows. Section 2 gives a brief introduction of Pregel,nal graph, but in those DSLs, there is no interface

and Section 3 introduce our high-level model andfor directly sending messages to other vertices, andthe DSL. Then, Section 4 sketches the transforma-

the only way to fetch non-local data is restrictedtion from our DSL to Pregel. Section 5 discusses

between neighbors in the original graph. For the

related work, and Section 6 concludes the paper.

same reason, the intention of modifying u’s parent’s
pointer in the else clause cannot be implemented in

2 Backgroundthose DSLs either, since u and u’s parent are not

We give a brief overview of the Pregel system [11]neighbors in the original graph, but modifying non-here.

local data relies on message passing as well. From
this example, we see that the existing DSLs are

2. 1 The Computation Modelnot expressive enough for implementing still a wide

In Pregel, the input is in general a directed graph,

range of algorithms.

and each vertex is associated with a mutable user-

To address these problems, we first propose in

defined state. The computation is split into super-

this paper a high-level model for vertex-centric

steps, and terminates when all vertices are inactive

computing on Pregel systems. During each stepin a particular superstep. A vertex is active at thebeginning of execution, but can deactivate itself byprocessors“voting to halt”. When inactive, it can only bevertex-centric
local
computation

activated by receiving external messages.
Within each superstep, the vertices compute inparallel, each executing the same user-defined function. A vertex can read the messages sent to it
in the previous superstep, modify its own state,and send messages to other vertices. Computa-

data
communication
barrier
synchronization

tion assumes the Bulk-Synchronous Parallel (BSP)
model [14], where global barrier synchronization

図2

The Pregel computation model

happens at the end of each superstep.
Vertices can communicate with each other by

Graph [5] and Mizan [8]. This paper does not tar-sending messages, each of which contains the ID

get a specific Pregel-like system. Instead, our DSLof the destination vertex and the message value. If

can be transformed to any Pregel-like system, asa vertex sends a message in superstep n, that mes-long as the system supports message passing be-sage will reach the destination vertex in superstep

tween arbitrary pairs of vertices, and provides then + 1. In general, a vertex can send any number of

mechanism of aggregators. In Section 4, we use

messages to any other vertex, but there is no guar-

physical supersteps, message passing and aggrega-

anteed order among the messages. In Pregel, only

tors to represent a Pregel program, and sketch howa single message type is allowed, so programmers

to transform our DSL to this internal representa-

should encode diﬀerent types in one if they need totion.

pass multiple types of messages.

3 The Domain-Specific Language

In addition, Pregel provides aggregators, a mech-

In this section, we introduce the concept of al-

anism for global communication. Every vertex cangorithmic superstep, and our domain-specific lan-provide a value to an aggregator in superstep n;guage for describing Pregel algorithms.

the system then combines these values by a reduction operator and makes the result available at

3. 1 The Algorithmic Superstep

the beginning of superstep n + 1. Pregel has a

In Pregel systems, one particular diﬃculty forbunch of predefined aggregators for various types,

programmers is to write vertex-centric programs

like max, min for numeric values, but program-in terms of supersteps and message passing. Ex-

mers may also define their own aggregators by pro-isting DSLs have already shown that thinking inviding the default value and a reduction operator.

terms of data fetching is more natural for Pregel

Whether build-in or user-defined, aggregation op-

programmers, because that allows algorithms to beerators should be commutative and associative.

expressed as more compact Pregel programs, fromwhich the necessary supersteps and message pass-

2. 2 Pregel-like Systems

ing can be automatically derived. However, it is

There are a bunch of open-source alternative toalso important to have the concept of superstep

the oﬃcial and proprietary Pregel system, suchin the algorithm level, otherwise it is hard to see

as Apache Hama [2], Apache Giraph [1], Catchwhether an algorithm can be properly parallelized

the Wind [13], GPS [12], GraphLab [10], Power-in Pregel. Therefore, the design of our DSL tries tofor describing algorithms for the following reasons.

(old graph)

data fetchingFirst, it tries to describe the most general intention of data fetching, that is, to get a field of another vertex whose ID is known. In this way, many
computations can be expressed by more naturally

old state

and concisely. Second, it allows a vertex to update
other vertices at the end of a superstep, which isalso a common and necessary pattern of communi-vertex-centriccomputation

cation in graph algorithms. Third, it clearly separates a superstep into three phases, which gives
programmers a standard template to describe the

new statealgorithms. The well-controlled side eﬀects and thehigh-level semantics make algorithmic supersteps
easier to be transformed into eﬃcient Pregel code.
Recall S-V PPA in Figure 1, which we can clearly(new graph)writing

describe in algorithmic supersteps. First, setting
all vertices as root nodes is simply a superstep that

図3

Algorithmic superstep

modifies every vertex’s pointer to its own ID. Second, within the iteration there is a single superstep,

give a proper abstraction of superstep, to let pro-in which a vertex fetches the pointers of its parent

grammers ignore unnecessary low-level details andand all its neighbors, and then modifies its own

retain the ability to reason about the algorithm.

pointer or the pointers of other vertices according

We model the basic unit of vertex-centric compu-

to the checking performed locally on that vertex.

tation as an algorithmic superstep, which contains
the following three phases (illustrated in Figure 3):

3. 2 Basic Vertex-Centric Programs• A local computation (LC) phase that pro-From now on, we propose a domain-specific lan-duces local results from the state of the cur-guage (DSL) for writing algorithmic supersteps

rent graph. The computation can fetch data

concretely. In our DSL, an algorithmic superstep isfrom any other vertices, either adjacent or non-described by a basic vertex-centric program, whichadjacent.

always starts with for (var in V), followed by a

• A local update (LU) phase that replaces the lo-

code block containing a list of statements, where

cal states of the current vertex with new values

var is a freely chosen name for the current ver-computed by the local computation phase.

tex. The syntax of basic vertex-centric programs is• A remote update (RU) phase that modifies thestates of other vertices.

shown in Figure 4.
This simple language contains the basic ele-

The LU and RU phases are optional, because aments in a normal imperative programming lan-vertex may not need to change local and remote

guage, such as assignment, branching, loop, andstates, but at least one of the two phases should bearithmetic operations. It also supports the basic

specified in an algorithmic superstep.

data types and data structures including integers,

The notion of algorithmic supersteps is better

booleans, floating-point numbers, pairs, and lists.

int
float
var
field

=
=
=
=

integerfloating-point number
identifier starting with lowercase letter
identifier starting with capital lettervcprog
block
stmt

::=
::=
::=
|
|
|
::=
|
|
|
|
|
::=
::=
|
::=
::=
::=
::=

for (var in V) block
stmt1 . . . stmtn (with same indentation)
if exp block else blockfor (var ← exp) block
var opassign exp
local | remote | agg
int | float | var | true | false
pair | list | fst exp | snd exp
var .Id | var .W (edge ID and weight)
exp opbinary exp | ( exp )(exp ? exp : exp)
opaccum list | fetch(exp, exp)
[ exp1 , . . . , expn ][ exp | var ← exp, exp1 , . . . , expn ]field(exp)new [ field ] oplocal exp
update [ fetch ] opremote exp
agg [ var ] opagg exp

current vertex.
Basically, fetching non-local data will introduce
additional communication with other vertices, andwith proper restrictions on fetching patterns, it

exp

pair
list
fetch
local
remoteagg
図4Syntax of basic vertex-centric program

There are four additional special constructs listed
at the end of Figure 4, whose semantics are closely
related to Pregel: the snapshot operator, the local

can be automatically transformed into Pregel. The
transformation will be discussed in Section 4.
Local Update.

A local update statement (rulelocal in Figure 4) is to update a local field by some
expression, where the operators are listed in Table 1.

At first glance, the local update statement
表1

Local Update Operators

Name

SyntaxDescription

simple assignmenta := ba←b

ADD assignment

a += ba←a+b

SUB assignmenta -= ba←a−b

MUL assignmenta *= ba←a×b

DIV assignmenta /= ba←a/b

AND assignment

a &&= ba←a∧b

OR assignmenta ||= ba←a∨b

MAX assignmenta >?= ba ← max (a, b)

MIN assignmenta <?= ba ← min(a, b)

update statement, the remote update statement,
and the aggregator.
Snapshot Operator.

The snapshot operator islooks similar to an assignment, but in our DSL it isto fetch data from local or non-local vertices, andactually diﬀerent. Recall the definition of algorith-

the value is the result of the previous logic super-

mic supersteps, in which the local update always

step. The syntax is simply field (exp), where thehappens after all local computation is completed.

field name is an identifier starting with a capital

To follow this specification, the local update state-

letter, and the expression specifies a vertex ID. Thisment can be seen as a deferred operation, which ex-

operator returns the value of the specified field ofpresses the intention of local data modification, andsome vertex, as the result of the previous algorith-

takes eﬀect after all local computation is finished.

mic superstep.

Local update statements can be put anywhere infor (u in V)
t := A(u) + B(C(u))

the program. Since our DSL is imperative, the fi-

In this example, suppose that A, B, and C are

example:fields, and that C stores the ID of another vertex.

for (u in V)
new[A] := B(u)new[B] := A(u)This piece of code calculates the sum of the current

nal result depends on the actual execution. Forvertex’s A field and vertex C(u)’s B field, and thenstores the result in the temporary variable t of theThis program swaps the values of fields A and B

of the current vertex, so B should be updated withsages are ordered in some specific way. Althoughcurrent A(u) and vice versa. In a traditional pro-it is possible for the receiver to order the update

gramming language, we usually have to use a tem-requests by adding an additional value in the mes-porary variable to achieve the swapping, but in oursage specifying the priority of execution, the price

DSL, we only need to write two local update state-is increasing the message size as well as having ad-ments, since local update is deferred.

ditional local computation. Besides, by investigat-

Remote Update.

The remote update state-

ing various practical Pregel algorithms, we have notment is to update a non-local field by some ex-seen the necessity of keeping the order of remotepression. The state to be modified is represented

updates.

by a snapshot operator, which specifies the field

Since the order of remote updates is not specified,name and the vertex, and the available assignment

as a consequence, it is not reasonable to use diﬀer-

operators are listed in Table 1 without the simple

ent operators in a basic vertex-centric program toassignment.

update a particular remote field.

The remote update statement (rule remote inAggregator.

Pregel aggregators are a mecha-

Figure 4) is also deferred. It expresses the inten-

nism for global communication, in which every ver-tion of remote update and takes eﬀect after all lo-

tex provides a value, before the Pregel system col-cal computation and update are performed. Morelects those values and use some reduction operator

precisely, an additional physical superstep is added

to produce a final value. Then, this result will be

for remote update, since communication is needed.

broadcast to all vertices in the graph, and will be

However, updating a local vertex and a remote ver-available in the next physical superstep.

tex are essentially diﬀerent, not only because of theSuch mechanism is provided in our DSL. In amessage passing, but also about the meaning of thebasic vertex-centric program, programmers can use

operation. When we modify a local field, we justthe rule agg in Figure 4, where each vertex evalu-

use a new value to replace the original one; how-ates the expression to a value, and the system re-ever, when we try to update a field of another ver-duces those values by a given operator. Currently,

tex, potentially there may be other vertices try-

we support all the default aggregators [11] in Pregel,ing to do the same.

Eventually, remote update

including maximum, minimum, sum, and , and or .

should be implemented by message passing (e.g.,

In addition, we provide a new aggregator random,

by sending the new value and the operation to an-which randomly selects an element according toother vertex), so in the updating phase, a vertex

a programmer-specified probability. In this case,may receive more than one update requests, which

each vertex provides a pair (vi , wi ) with wi ≥ 0,

is diﬀerent from the local modification. To makeand the probability that vi is selected as the final∑
result is wi /( n−1i=0 wi ).

programmers aware of this situation, we forbid a
remote update statement to be a normal assignment. Instead, programmers should consider how

3. 3 Concatenation and Iteration

to produce a final result from the original value andIn the previous subsection, we have presented thea (possibly empty) list of new values.

design of basic vertex-centric programs, with which

Unlike the imperative local update statements,

we can access local or non-local data, do pure local

we do not specify the order of performing remotecomputation, and modify the fields on related ver-updates, since Pregel does not guarantee that mes-

tices. In this subsection, we introduce how to write

prog

term

図5

::=
|
|
::=
|
|

vcprogprog1 . . . progn (with same indentation)
do exp until term
iter int
fix [ field1 , . . . ,f ieldn ]exp (no local variables)

Syntax of concatenation and iteration

D field, which stores the pointer. Next is an iteration containing a single basic vertex-centric program, which tries to find new neighboring pair ofvertices that are not in the same tree yet. In thisprogram, we first see whether u’s parent is a root,by checking the condition D(D(u)) == D(u).

Ifthis condition is satisfied, it fetches all its neighsophisticated algorithms using two combinators on

bors’ parents and chooses the ones whose ID isbasic vertex-centric programs: concatenation andsmaller than u’s parent, which is represented by

iteration.

the list comprehension. Then, the build-in function

Concatenation combines two vertex-centric pro-minimum further picks out the one with the small-grams into one sequentially, so that the second pro-

est ID among them and updates its parent D(u)’s D

gram can take the result of the first program as

field. The operator used here is <?=, which means if

input. Iteration, on the other hand, runs a vertex-

the provided value is smaller than D(D(u))’s orig-

centric program repeatedly until some termina-

inal value, then perform the assignment. For thetion condition is satisfied. In our DSL, program-other case, where u’s parent is not a root, we justmers can enclose a vertex-centric program with

update the local field D with its grandparent. The

do . . . until term, and three types of termination

algorithm terminates when all vertices’ pointers no

conditions can be used:

longer change.

1. Iterate a fixed number of supersteps.
2. Iterate until the values of some fields do notchange anymore.

4 Transformation to PregelIn this section, we sketch how to transform our

3. Iterate until an expression evaluates to true,

DSL to physical supersteps in Pregel. We first con-

where the expression should not refer to local

sider the transformation of basic vertex-centric pro-variables and fields.

grams in Section 4. 1, and then concatenation anditeration in Section 4. 2. Finally, in Section 4. 3, we

3. 4 Example: Shiloach-Vishkin Practical
Pregel Algorithm

show concretely how the SV-PPA is transformed toPregel.

Here, we give the complete implementation of
SV-PPA (Figure 1):

4. 1 Basic Vertex-Centric Programfor (u in V)
new[D] := udofor (u in V)
if D(D(u)) == D(u)update[D(D(u))] <?= minimum[ D(e.Id)| e <- Nbr(u), D(e.Id) < D(u) ]else
new[D] := D(D(u))
until fix[D]Let us first consider how a basic vertex-centric
program can be translated into a sequence of physical supersteps. In general, a basic vertex-centric
program will be transformed into several supersteps
for data fetching, one superstep for main computing and local state updating, and then one superstep for remote updating. The main challenge ofthe transformation lies in the data fetching phase,In the first two lines, we write an initialization

where some other vertices may be involved and playprogram which assigns each vertices’ own ID to its

the role of data provider, and some data communi-

cation is needed; this will take most of this subsec-

request to the vertex whose ID is the result of eval-tion to explain.

uating exp. In superstep n + 1, every vertex replies

Data fetching.

In the DSL, data fetching is

to all requests. Next, in superstep n + 2, every ver-specified by the snapshot operator, in which thetex gets the results, and then recreates the results

source vertex and the field are specified. The strat-

for further computation.

egy is a query-reply conversation, where a vertex

There is a special case where l is actually a list

first sends a message to the source to acquire theof edges in the graph, and by evaluate exp we findfield, and then the source vertex replies the valuethat we need to fetch some field alone the edges.

of that field. For example, suppose that the expres-In this case, we do not need to send request to allsion has the form field (exp), and that the current

neighbors. Instead, we just send the vertex’s field

vertex u (potentially recursively) evaluates exp to

directly to neighbors, which saves one superstep.

a vertex ID x in superstep n, and that the field we

In the transformation procedure above, we haveneed is A. Then, u sends a message to vertex x in

only shown how to transform each snapshot into

superstep n including its own ID, and attaches aseveral steps of query-reply conversation, but fi-

unique label l1 to that message. At the beginning

nally we should combine them together, so that dif-of superstep n + 1, all vertices scan the messages,

ferent data fetching procedures can actually happen

and then reply each request having label l1 with thein parallel, and the code in some supersteps should

value of its own field A to the sender whose ID is

be merged. When to perform each data fetching

included in that request, and also attaches a uniqueprocedure is decided by a dependency check, whilelabel l2 to the reply. In superstep n + 2, u scans

combining code in supersteps for fetching is always

the messages and find the one with label l2 , whichsafe, because all communications use diﬀerent chan-should contain the value of field (exp).

nels by attaching unique labels to messages. Since

When the snapshot operator contains a loop vari-

we separate a single program into multiple super-able (from either a for statement or a list com-steps, some local variables may be referred to by theprehension), the situation becomes slightly morecode in diﬀerent supersteps. In this case, we should

complicated. In our DSL, a restriction is that ascan the program one more time, and change such

snapshot may only contain a loop variable from akind of local variables into local fields, so that they

top-level for statement or list comprehension; oth-can be stored on the vertex.

erwise the operation is considered invalid.

Local and remote updates.

One

Finally, trans-

reason for this choice is that, a use of the snap-

forming the local update and remote update is triv-

shot operator that contains multiple loop variables

ial. In general, we should store the new value intois very expensive, which may cause a large num-the vertex state, and then add an assignment in

ber of messages passing through the system, andthe local update superstep after all local compu-

may be also space consuming. If, however, we only

tation is done. Then, a remote update statement

consider the snapshot operator in a loop containing

should be translated to a message sending contain-

no inner loops, then the translation is easy. Since

ing the ID of the destination vertex and the value.

the loop is represented by iterating over a certain

Since a remote update statement in our DSL does

list (either a for loop or list comprehension), we

not necessary happen just before the remote up-

first evaluate the list as l, and then in superstep n,date superstep, in which case we should also keep

we evaluate the exp and create a loop to send thethe information in the vertex state, and send themessages before the remote update phase. Finally,in the remote update phase, the only thing to do isreading the update requests, and update the corresponding fields of the vertex.

In these equations, the ++ operator combines twogroups of supersteps such that the last superstep inthe first group and the first superstep in the second
group are merged together. The resulting program

4. 2 Concatenation and Iteration
In the previous subsection, we have shown how totransform a basic vertex-centric program into physical supersteps. Here, we introduce the transformation rules for concatenation and iteration, and try
to derive a complete Pregel program with as small
number of supersteps as possible.

also contains the three parts. It is obvious thatthe rules defined above are correct, because the order of supersteps in those two programs are kept
in the combined program, and the order of the two
programs are also kept.
Secondly, we show the transformation rules for
iteration with three diﬀerent types of termination

A basic vertex-centric program may possibly betransformed into several sending supersteps, one
computing superstep, and possibly one remote update superstep. The sending supersteps are purein the sense that no local state would change; thesingle computing superstep may update the local
state; and in the remote update superstep, a vertex only scans the messages and updates its own

conditions. Each type of termination rule can beeasily translated to a 4-tuple (I, T, J, E), where I
is an initialization superstep, T is a superstep containing termination check, J is to generate global
information about whether to exit the iteration andjump to superstep T , and E is the exit superstep
doing nothing. The transformation rules for iteration is as follows:state. In our transformation for concatenation anditeration, we use a slightly diﬀerent representation,=in which a program may possibly have the firstsending superstep, followed by other sending supersteps and the computation superstep, and an
optional remote update superstep.

For simplic-===iterate((I, T, J, E), [ F S, SC, RU ])[ ∅, [ F S ++ I, T ++ SC, RU ++ FS ++ J ], E ]iterate((I, T, J, E), [ F S, SC, ∅ ])[ ∅, [ F S ++ I, T ++ SC ++ FS ++ J ], E ]iterate((I, T, J, E), [ ∅, SC, RU ])[ ∅, [ I, T ++ RU ++ SC ++ J ], E ]iterate((I, T, J, E), [ ∅, SC, ∅ ])[ ∅, [ I, T ++ SC ++ J ], E ]ity, we represent the program in a concise way like
[F S, SC, RU ], where F S is the first sending super-Now let us see how the three types of termina-

step, SC is the other sending supersteps along withtion rules can be translated to an (I, T, J, E) tu-

the local computation superstep, and RU is theple. First is the condition of iterating a fixed num-

remote update superstep. Since F S and RU are

ber of times. This requires an additional field asoptional, we simply write ∅ when they are omit-

the counter.

ted. The following equation shows the concatena-

the counter to 0, and the termination check steption rules for combining two programs:

T reads the counter, and goes to superstep J (by

=
=
=
=

concat([ F S1 , SC1 , ∅ ], [ ∅, SC2 , RU2 ])[ F S1 , [ SC1 ++ SC2 ], RU2 ]concat([ F S1 , SC1 , ∅ ], [ F S2 , SC2 , RU2 ])[ F S1 , [ SC1 ++ F S2 , SC2 ], RU2 ]concat([ F S1 , SC1 , RU1 ], [ ∅, SC2 , RU2 ])[ F S1 , [ SC1 , RU1 ++ SC2 ], RU2 ]concat([ F S1 , SC1 , RU1 ], [ F S2 , SC2 , RU2 ])[ F S1 , [ SC1 , RU1 ++ F S2 , SC2 ], RU2 ]The initialization step I initializes

changing the global state) if the counter is greaterthan a pre-determined value. In superstep J, we
just increase the counter by one. Second is thecondition of iterating a program until some specified fields remain unchanged during the iteration.
In this case, we need to trace all involved fields,iteration. In the iteration, we can see that the main
step1

u

initialization
D (u) ← urequestD ( D (u))

D (u): parent node
M (u): u modified or not
Magg : any vertex modified

data fetching can be done. There are two non-local
Magg ← ORu∈V true

step2

u

data sending

step3

D (u)

du

requestsv1

if (not Magg )goto step5
reply
D 0 (u) ← D (u) D ( D (u))

u

v1

du

· · · vnu’s neighbours

dvi

remote update
D ( D (u)) ← mini dvi
M(u) ← D (u) 6= D 0 (u)D 0 (u) ← D (u)requestD ( D (u))
Magg ← ORu∈V M (u)

step5

u

termination

expressions that are required by u’s computation:
the parent’s pointer D(D(u)) and some neighbor’s
pointer D(e.Id ). The parent’s pointer is a simple

D ( vi )u· · · vnu’s neighbours

local update
if D ( D (u)) = D (u)if D ( D (u)) 6= D (u)dv ← mini D (vi )D (u) ← D ( D (u))
if (dv < D (u)) send dv

step4

computation happens in Step 3, and before that, alldupattern that fetches data from a single source D(u),but the fetching of neighbor’s pointer is inside a list
comprehension, so they are transformed diﬀerently.
For D(D(u)), u sends the request to D(u) in Step
1, and the parent replies in Step 2, so that all vertices can get D(D(u)) in Step 3. Then, for D(e.Id )(in step 2)

voteToHalt()

in the list comprehension expression, we can seethat it is passed along the edges, according to the

図6

S-V PPA transformation result

generator e ← Nbr(u). To translate this pattern,
we simply let every vertex send their D(u) to allso we insert code to keep a copy of old values intheir neighbors in Step 2, so that in Step 3, ev-

superstep T which is prior to any state modifica-

ery vertex will receive D(vi ) from all its neighborstion, so that in the end of iteration J, every ver-{v1 , . . . , vn }. After all data have been sent, thetex knows whether its own state is modified duringcomputation is performed in Step 3: Every vertex

the iteration. Besides, in superstep J, the vertices

first check whether its parent is a root node, by

use global communication (aggregator in Pregel)checking the equation D(D(u)) = D(u). Then, it

to obtain a global variable indicating whether any

either performs a local update in Step 3, or performof the vertices is modified in an iteration, so thata remote update, which happens at the beginningin superstep T , the vertices look at this variableof Step 4. Next, in Step 4, we trace the D field

to decide whether to exit the iteration. Third isto see whether it changes in the iteration. We use

the termination condition using a general expres-an aggregator to generate a global value Magg in-sion that contains no local variable. In this case,dicating whether there is any modification, so we

we simply evaluate the expression in superstep J,

jump to Step 2, check the value of Magg , and decide

and then perform checking in T .

whether to exit the iteration. Finally we reach Step
5, where we just invoke the function voteToHalt().

4. 3 Example: Shiloach-Vishkin Practical
Pregel Algorithm

In this figure, each intention of data fetchingis transformed separately, but finally they can be

Now let us see concretely how S-V PPA is trans-

easily put together, since there is no vertex state

lated to Pregel’s physical supersteps. We visualizechange in the data fetching phase, and the message

the transformation result in Figure 6.

types can be distinguished by adding an additional

First of all is the initialization step, in which alllabel. In implementation, we just combine the code

vertices set their pointers (D field) to their own

in each superstep, handle the messages properly,

IDs. Then, M is an additional boolean field, whichand perform the state transition at the end of theis used to trace whether D changes in the following

superstep.

参 考 文 献

5 Related Work
There are lots of DSLs proposed for big graphprocessing on Pregel. Fregel [3] is a functional language for writing Pregel algorithms, in which thevertex-centric computation is represented by purefunctions returning new vertex state.

However,

it cannot access data on non-adjacent vertices, ormodify the state of other vertices. Green-Marl [6] isan imperative language for shared-memory multiprocessor system, and can be compiled to Pregel[7], but the compilation relies on discovering Pregel-canonical pattern, which means that programmers
may easily get compilation errors if they are notfamiliar with the implementation of the compiler.
It also has the same expressiveness problem as
Fregel. Palovca [9] is an embedded domain specificlanguage using Haskell as the underlying host language for Pregel algorithms. It uses a monad tohide side eﬀects and define vertex-centric programin a low-level way like in typical Pregel systems.

6 Conclusion
This paper introduces the concept of algorithmic
superstep, a high-level model for Pregel computation, and presents a novel domain-specific languagethat allows programmers to implement Pregel algorithms in a more natural way.

Our DSL has

high expressiveness that can represent complicatedPregel algorithms concisely, and the high-level semantics makes it much easier to be transformed toPregel with small number of supersteps. As future
work, we are interested in the further optimizationof reducing the number of supersteps and the sizeof internal data, to produce more eﬃcient Pregel
programs.
Acknowledgment We thank Dr. Kento Emoto(from Kyushu Institute of Technology) for his insightful comments to improve this work.

[ 1 ] : Apache Giraph, http://giraph.apache.org/.
[ 2 ] : Apache Hama, http://hama.apache.org/.
[ 3 ] Emoto, K., Matsuzaki, K., et al.: Think like avertex, behave like a function!, Proceedings of ACM
SIGPLAN International Conference on Functional
Programming, ACM, 2016.
[ 4 ] Gabow, H. N. and Tarjan, R. E.: A linear-timealgorithm for a special case of disjoint set union,
Journal of computer and system sciences, Vol. 30,
No. 2(1985), pp. 209–221.
[ 5 ] Gonzalez, J. E., Low, Y., et al.: Powergraph:
distributed graph-parallel computation on naturalgraphs, Presented as part of the 10th USENIX Symposium on Operating Systems Design and Implementation, 2012.
[ 6 ] Hong, S., Chafi, H., et al.: Green-Marl: a
DSL for easy and eﬃcient graph analysis, ACM
SIGARCH Computer Architecture News, ACM,
2012, pp. 349–362.
[ 7 ] Hong, S., Salihoglu, S., Widom, J., and Olukotun, K.: Simplifying scalable graph processing
with a domain-specific language, Proceedings ofAnnual IEEE/ACM International Symposium on
Code Generation and Optimization, ACM, 2014,
pp. 208.
[ 8 ] Khayyat, Z., Awara, K., et al.: Mizan: a system for dynamic load balancing in large-scale graphprocessing, Proceedings of the ACM European Conference on Computer Systems, 2013.
[ 9 ] Lesniak, M.: Palovca: describing and executing
graph algorithms in haskell, International Symposium on Practical Aspects of Declarative Languages,
2012.
[10] Low, Y., Bickson, D., et al.: Distributed
GraphLab: a framework for machine learning anddata mining in the cloud, Proceedings of the VLDB
Endowment, Vol. 5, No. 8(2012), pp. 716–727.
[11] Malewicz, G., Austern, M. H., et al.: Pregel: asystem for large-scale graph processing, Proceedingsof the ACM SIGMOD International Conference on
Management of Data, 2010.
[12] Salihoglu, S. and Widom, J.: GPS: a graphprocessing system, Proceedings of the International
Conference on Scientific and Statistical Database
Management.
[13] Shang, Z. and Yu, J. X.: Catch the Wind: graphworkload balancing on cloud, Proceedings of the International Conference on Data Engineering, 2013.
[14] Valiant, L. G.: A bridging model for parallel
computation, Communications of the ACM, (1990),
pp. 103–111.
[15] Yan, D., Cheng, et al.: Pregel algorithms
for graph connectivity problems with performance
guarantees, Proceedings of the VLDB Endowment,(2014), pp. 1821–1832.