Programming and Debugging Large- Scale Data Processing Workflows

(1)

Programming and Debugging Large-‐

Scale Data Processing Workﬂows

Christopher Olston

Google Research

(2)

Context

• _{Elaborate
processing
of
large
data
sets}

e.g.:

• web search pre-‐processing

• cross-‐dataset linkage

• web informaKon extracKon

serving

inges)on

storage &

_processing

(3)

Context

Debugging aides:

• Before: example data generator

• During: instrumentaKon framework

• ANer: provenance metadata manager

storage &

processing

scalable ﬁle system

e.g. GFS

distributed

sorKng & hashing

e.g. Map-‐Reduce

dataﬂow programming

framework

e.g. Pig

workﬂow manager

e.g. Nova

Detail:

Inspector

Gadget,

RubySky

Overview

(4)

Pig: A High-‐Level Dataﬂow

Language & RunKme for Hadoop

Visits = load ‘/data/visits’ as (user, url, time);!

Visits = foreach Visits generate user, Canonicalize(url), time;!

!

Pages = load ‘/data/pages’ as (url, pagerank);! !

VP = join Visits by url, Pages by url;! UserVisits = group VP by user;!

Sessions = foreach UserVisits generate flatten(FindSessions(*));!

HappyEndings = filter Sessions by BestIsLast(*);!

!

store HappyEndings into '/data/happy_endings';!

(5)

vs. map-‐reduce: less code!

0 20 40 60 80 100 120 140 160 180

Hadoop

Pig

1/20 the lines of code

0 50 100 150 200 250 300

Hadoop

Pig

Minutes

1/16 the development Kme

performs on par with raw Hadoop

"The [Hofmann PLSA E/M] algorithm was implemented in pig in 30-35 lines of pig-latin statements. Took a lot less compared to what it took in implementing the algorithm in Map-Reduce Java. Exactly that's the reason I wanted to try it out in Pig. It took 3-4 days for me to write it, starting from learning pig.” "

"

(6)

vs. SQL:

step-‐by-‐step style;

lower-‐level control

"I much prefer writing in Pig [Latin] versus SQL. The step-by-step method of" creating a program in Pig [Latin] is much cleaner and simpler to use than the single block method of SQL. It is easier to keep track of what your variables are, and where you are in the process of analyzing your data.” "

"

-- Jasmine Novak, Engineer, Yahoo!"

"PIG seems to give the necessary parallel programming construct (FOREACH, FLATTEN, COGROUP .. etc) and also give sufficient control back to the

programmer (which purely declarative approach like [SQL on Map-Reduce] doesnʼt).” "

"

(7)

Transform

to (user, Canonicalize(url), Kme)

Load

Pages(url, pagerank) Load

Visits(user, url, Kme)

Join url = url Group by user Transform

to (user, Average(pagerank) as avgPR) Filter

avgPR > 0.5

Conceptually:

A Graph of Data TransformaKons

(8)

Transform

Join url = url

Group by user

Transform

to (user, Average(pagerank) as avgPR)

Filter avgPR > 0.5

Load

(Amy, 0.65) (Amy, 0.65) (Fred, 0.4)

(Amy, { (Amy, www.cnn.com, 8am, 0.9), (Amy, www.snails.com, 9am, 0.4) })

(Fred, { (Fred, www.snails.com, 11am, 0.4) }) (Amy, www.cnn.com, 8am, 0.9)

(Amy, www.snails.com, 9am, 0.4) (Fred, www.snails.com, 11am, 0.4) (Amy, cnn.com, 8am)

(Amy, hip://www.snails.com, 9am) (Fred, www.snails.com/index.html, 11am)

(Amy, www.cnn.com, 8am) (Amy, www.snails.com, 9am) (Fred, www.snails.com, 11am)

(www.cnn.com, 0.9) (www.snails.com, 0.4)

Ill

us

tr

ate

d!

“ILLUSTRATE lets me check the output of my lengthy batch jobs and their custom functions without having to do a lengthy run of a long pipeline. [This feature] enables me to be productive.” "

"

(9)

Transform

Join url = url

Group by user

Transform

to (user, Average(pagerank) as avgPR)

Filter avgPR > 0.5

Load

(Amy, cnn.com, 8am)

(Amy, hip://www.snails.com, 9am) (Fred, www.snails.com/index.html, 11am)

(Amy, www.cnn.com, 8am) (Amy, www.snails.com, 9am) (Fred, www.snails.com, 11am)

(www.youtube.com, 0.9) (www.frogs.com, 0.4)

(N

aï

ve

A

lg

or

ith

m

)

(10)

(11)

InteracKve Graphics Programming UI

by Bret Victor, Apple Designer (2012)

“Creators need an immediate connection to what they create … If you make a change … you need to see the effect of that immediately.” "

"

(12)

Pig Today

• Open-‐source (the Apache Pig Project)

– 

Dev./support/training by Cloudera, Hortonworks

– 

Oﬀered on Amazon ElasKc Map-‐Reduce

• Used by LinkedIn, Neqlix, Salesforce, Twiier, Yahoo ...

• At Yahoo, as of early 2011:

– 

1000s of jobs/day

– 

75%+ of Hadoop jobs

• Has an interacKve example-‐generator command,

(13)

✔

Next: NOVA

Debugging aides:

• Before: example data generator

• During: instrumentaKon framework

• ANer: provenance metadata manager

storage &

processing

scalable ﬁle system

e.g. GFS

distributed

sorKng & hashing

e.g. Map-‐Reduce

dataﬂow programming

framework

e.g. Pig

workﬂow manager

e.g. Nova

✔

(14)

Why a Workﬂow Manager?

• _{ConKnuous
data
processing
(simulated
on
top}

of Pig/Hadoop staKc processing layer)

(15)

Example

Workﬂow

template detecKon news site templates news arKcles RSS feed template tagging shingling de-‐ duping unique arKcles ALL NEW NEW NEW NEW NEW NEW NEW ALL ALL shingle hashes seen ALL NEW

(16)

Data Passes Through

Many Sub-‐Systems

GFS

Map-‐Reduce

Pig

Nova

low-‐latency

processor

serving

ingesKon

datum X

datum Y

metadata

queries

provenance of X?

(17)

Ibis Project

• Beneﬁts:

– 

Provide uniform view to users

– 

Factor out metadata management code

– 

Decouple metadata lifeKme from data/subsystem lifeKme

• Challenges:

– 

Overhead of shipping metadata

– 

Disparate data/processing granulariKes

data processing sub-‐systems

metadata manager

users

metadata

queries

answers

metadata

Ibis

integrated metadata

(18)

Example Data and Process

GranulariKes

Pig script

Pig job

Pig logical

operaKon

MR job

Pig physical

operaKon

MR job phase

MR task

Task aiempt

data granulari7es

process granulari7es

Table

Column

group

Row

_Column

Cell

AlternaKve

Web page

Workﬂow

MR program

(19)

What’s Hard About

MulK-‐Granularity Provenance?

• _{Inference:
Given
relaKonships
expressed
at}

one granularity, answer queries about other

granulariKes (the seman;cs are tricky here!)

• _{Eﬃciency:
Implement
inference
without}

resorKng to materializing everything in terms

of ﬁnest granularity (e.g. cells)

(20)

)tle year lead actor

Avatar 2009 A1: Worthington

A2: Saldana

IncepKon 2010 DiCaprio

Avatar 2009 Saldana IncepKon 2010 DiCaprio

Avatar 2009 Worthington IncepKon 2010 DiCaprio Yahoo! Movies web page IMDB web page map output 1 map output 2 Yahoo extracted table

IMDB extracted table

combined table

Inferring TransiKve Links

Q: Which web page said “Saldana”?

map phase

(21)

✔

Next: INSPECTOR GADGET

Debugging aides:

• Before: example data generator

• During: instrumentaKon framework

• ANer: provenance metadata manager

storage &

processing

scalable ﬁle system

e.g. GFS

distributed

sorKng & hashing

e.g. Map-‐Reduce

dataﬂow programming

framework

e.g. Pig

workﬂow manager

e.g. Nova

✔

(22)

MoKvated by

User Interviews

• _{Interviewed
10
Yahoo
dataﬂow
programmers}

(mostly Pig users; some users of other

dataﬂow environments)

(23)

Summary of User Interviews

# of requests feature

7 crash culprit determinaKon

5 row-‐level integrity alerts

4 table-‐level integrity alerts

4 data samples

3 data summaries

3 memory use monitoring

3 backward tracing (provenance)

2 forward tracing

2 golden data/logic tesKng

2 step-‐through debugging

2 latency alerts

1 latency proﬁling

1 overhead proﬁling

1 trial runs

(24)

Our Approach

• _{Goal:
a
programming
framework
for
adding}

these behaviors, and others, to Pig

• _{Precept:
avoid
modifying
Pig
or
tampering}

with data ﬂowing through Pig

• _{Approach:
perform
Pig
script
rewriKng
–}

(25)

group

count

join

ﬁlter

load

IG

coordinator

store

IG agent

(26)

Example:

Integrity Alerts

alert!

propagate alert

to user

group

count

join

ﬁlter

load

IG

coordinator

store

IG agent

(27)

Example:

Crash Culprit Determina;on

record counts

Phases 1 to n-‐1:

records

Phase n:

Phases 1 to n-‐1:

maintain count lower bounds

Phase n:

maintain last-‐seen records

group

count

join

ﬁlter

load

IG

coordinator

store

IG agent

(28)

Example:

Forward Tracing

tr

ac

in

g in

str

uc

Ko

ns

report traced

records to user

group

count

join

ﬁlter

load

IG

coordinator

store

IG agent

traced records

(29)

join

ﬁlter

load

IG

coordinator

IG agent

store

dataﬂow engine run;me

applicaKon

launch instrumented dataﬂow run(s) end user dataﬂow program + app. parameters

IG driver library

raw result(s) result

(30)

Agent & Coordinator APIs

Agent Class

init(args)

tags = observeRecord(record, tags)

receiveMessage(source, message)

ﬁnish()

Coordinator Class

init(args)

receiveMessage(source, message)

output = ﬁnish()

Agent Messaging

sendToCoordinator(message)

sendToAgent(agentId, message)

sendDownstream(message)

sendUpstream(message)

Coordinator Messaging

sendToAgent(agentId, message)

(31)

ApplicaKons Developed For IG

# of requests

feature

lines of code (Java)

7 crash culprit determinaKon

141

5 row-‐level integrity alerts

89

4 table-‐level integrity alerts

99

4 data samples

97

3 data summaries

130

3 memory use monitoring

N/A

3 backward tracing

(provenance)

237

2 forward tracing

114

2 golden data/logic tesKng

200

2 step-‐through debugging

N/A

2 latency alerts

168

1 latency proﬁling

136

1 overhead proﬁling

124

(32)

RubySky: ScripKng Framework with

Built-‐in Debugging Support

• _Scenario:

_ad-‐hoc

_{data
analysis}

– 

e.g. we just did a crawl; is it too biased toward

Yahoo content?

• _{In
this
scenario,
someKmes
using
new
data;}

always running new code

à Bugs are the norm

(33)

RubySky Example

:Crawl_data <= load(context, "/crawl_data", schema {string :url; ...})! !

# quick-and-dirty determination of "domain" from "url" ! # (e.g. http://www.yahoo.com/index.html => yahoo.com)!

:With_domains <= Crawl_data.foreach() do |r|! url = r[:url]!

prefix = “http://www.”!

if (url[0, prefix.len] == prefix)!

domain = url[prefix.len, url.index('/', prefix.len) - prefix.len]! else!

throw “Don’t know how to parse this url: “ + url!

end! Tuple.new([url, domain])! end! ! :Yahoo_only <= With_domains.filter do |r|! r[:domain] =~ /^yahoo/! end! ! :Grouped <= Yahoo_only.group! !

(34)

RubySky Example:

w/debug

stmts

...!

:With_domains <= Crawl_data.foreach() do |r|! ...!

else!

domain = PUNT url!

end!

Tuple.new([url, domain])! end!

!

# examine a few extracted domains, as sanity check!

:Domain_sample <= With_domains.foreach() do |r|! $count += 1!

($count < SAMPLE_SIZE)? [Tuple.new([r[:domain]])] : []! end!

SHOW Domain_sample!

!

# alert if any extracted domain is empty or null!

:Domain_missing <= With_domains.foreach() do |r|! (r[:domain] == "" or r[:domain] == nil)? !

Programming and Debugging Large- Scale Data Processing Workflows