Original subtitles

Hi everyone, welcome to the HINTS video for Project 2.3, Iterative Machine Learning Training.

In this video, I will walk you through the following topics.

First, let's take a look at the starter code in DABS.

We'll figure out the format of the samples RDD and the rationale behind using sparse format.

Then we'll focus on the scalability problem of the starter code

and think about how we can store weights array in the worker nodes rather than purely in the driver node.

And I will give the first hint about how to use inverted index and join-based communication.

Then we will talk about the performance problem of the draft optimization

and how to utilize partitions to optimize join operators.

I will give an introduction for RDD join operators after that.

At last,

I will give some further tips that we want to share with you guys.

Yeah.

OK, so let's get started.

As you have watched in the overview video,

this is the core logic of our machine learning training.

We have data samples and model parameters as inputs.

For data samples, it is stored as an RDD and it's also partitioned across executors.

For model parameters, it is stored as a weight array in the driver.

Next is the training iteration.

Specifically, the driver will broadcast the weights to all executors.

Then each executor will compute the gradient updates with respect to the data partition that resides in the executors.

And the driver would collect updates from executors and then update the parameters locally.

Then the iteration will repeat until the model converges.

Here we focus on the most significant computation, namely the computation of gradient descent.

So for gradient descent, we have broadcasted weight array and the worker's local samples as input.

We need to compute prediction based on the result of the sigmoid function.

But before that, we need to do a dot product between the broadcasted weight array and the feature values for each sample.

Then we can use the prediction result to compute the gradient and of course the loss function.

As we can see here, since the dot product will combine values from both inputs,

therefore it's the most important computation that we will consider for optimization.

Let's go through an example.

Here we have five values, oh sorry, five features.

Therefore the weight array also has five elements and the sample also got five.

By dot producting them, we can get a result.

However, it's clear that since there are a lot of zeros, as you can see here in each sample,

the final result also contains a lot of zeros.

Which means if we use the whole sample array to do dot product,

then there will be a lot of unnecessary computation.

This is especially true in real-world training sets,

since there are typically a lot of zeros in those samples,

or null values in those samples.

So how can we solve this problem?

The answer is using sparse format instead of dense format.

Rather than storing the whole dense vector for each sample,

we store those feature values that are non-zero.

And here is a logic representation for this.

For sparse format, we have a key-value pair list for each sample,

and the key is the feature ID while the value is the feature at value.

For example, here 0 represents its feature 0,

and 1 means feature 0 has value equals to 1.

Note that for the starter code,

the physical representation is actually a tuple,

with three elements,

a label,

a feature ID list,

and a feature value list.

Then, in order to dot product,

we need to use a feature ID list

to retrieve all the corresponding weights

and do the dot product

between them and the feature value list.

Through this way,

those computations for zero,

for zero products,

can be saved.

As a quick summary,

the final samples RDD looks like this.

For each sample,

we will only store useful features

and their corresponding non-zero values.

Next,

we will take a look at the scalability problem in the starter code.

From a high level,

there are two limitations in the starter code.

First,

we need a single-node memory.

The training set might have too many features

to fit into a single-node memory.

Secondly,

for communication,

since each partition might only require

a subset of parameters to do the dot product,

then broadcasting the whole weight array across the nodes

might be a waste.

So,

how can we improve the implementation?

The first step is to partition the weight array,

like what we did for the sample data,

so that we can store them across the whole cluster

instead of only within one single node.

Then,

we need to figure out a way

to tag each weight with a feature ID.

After this step,

we can get the weights ready,

which the key is a feature ID

and the value is a feature weight.

The next step is that we need to bring the feature weights

to each sample

so that we can do the dot product

and further computation.

In general,

the goal is to combine samples RDD

and the weights RDD together.

A simple way to do this

is to use RDD lookup operation

for each sample.

Since we know its relevant feature ID,

we can use those feature IDs

to fetch each weight with RDD

and they can do the computation.

However,

since lookup can only bring weight

to the driver node,

we still need to either broadcast weight

to worker node

or fetch each sample to the driver,

then do the computation.

This is very slow.

So,

is there any better way to do the combination?

Actually,

what we want here is

a distributed shuffle operation

to combine the samples dataset

and the weights dataset.

And let's use the join operation.

One potential issue is that

the weights already

sample

is a KV pair,

but samples RDD is a list.

To solve this,

we will consider inverted index.

Here is an inverted index computation pipeline

for samples RDD.

As shown here,

we can first do a map operation

to construct multiple key-value pairs

for each sample

as shown in the operation here,

where key is a feature ID

and the value is a

single-sample sparse vector.

Note that

since different samples might have

the same relevant feature value,

for example,

sample 2 will generate a pair for

feature 3,

and sample 3 would also generate

a key-value pair

for the same feature,

that is, feature 3.

So,

we will

as a

we can use the

groupByKey operation

to group those

values

with the same key together,

so that we don't have like duplicate keys

specifically.

Here, for example,

for feature 3,

the values would be a list

that contains

sample 2 and sample 3.

And at last,

we'll have a key-value data set

mapping from feature ID

to a list of samples.

Now,

we can do joint operations

between the intermediate data set

with the weights RDD

because they own the

same keys.

So,

here is the draft data flow

for our training task.

As shown in this whole pipeline

after we do the joint operation,

we will get a

new mapping

that is from

feature ID

to its weight value

and the list of

samples

here.

Note that for each key-value pair,

we only have one feature.

So,

we're still not able

to do the dot product here.

Therefore,

in order to

get each sample

to get all its relevant parameter weights,

we can do an

inverted index again.

That is,

to construct a new data set

with samples as the key

while

a whole relevant weight list

as the value.

By this way,

each sample can update itself

based on the corresponding

weight list.

Unfortunately,

life is not easy.

You might have noticed

the warning sign here.

There are several problems

for this design.

For one thing,

the first inverted index

is very expensive.

It's possible that

we replicate the whole

sample data set

multiple times.

For example,

if we got a hot feature,

then every sample

that has a non-zero value for it,

then for the feature's

key-value pair,

we have to store the whole data set

in one list,

which is clearly not acceptable.

For another,

the join operation

is also very expensive,

since the number of features

might be huge,

in which case,

we are trying to join

a large number of records

while each pair of weights

has a very small size,

namely one single feature weight,

so it is also not desirable.

Before we give

our next thing,

after this part,

you should be able

to calculate

and know what is

the inverted index

using RDD operations

and know the basic ideas

of how to do

join-based communication

to combine two RDDs.

Make sure you comprehend those

and then we can proceed

to the next steps.

So now we focus on

the performance part.

As we stated before,

one of our

jacked implementation

drawbacks is that

it tries to join

a large number of records,

and each record

is small in size.

From the test case perspective,

for example,

for test case A,

it contains over

20 million features,

which makes the join cost

unacceptable

because it joins

on the feature side.

Let us now

take a look at

a test case

that has multiple times

of the feature dimensions

compared to test case A.

So our goal is that

we want to join

a smaller number of records

while each of them

has a larger size.

And we want to avoid

storing multiple copies

of each sample.

Okay,

so here is our second hint.

Instead of using

logical features

or sample IDs,

let's use partition ID.

And based on the partition ID

to do the join operation.

In this way,

we can adjust

the number of partitions

so that we can control

the size of the

to-be-joined record.

So we have a partition ID

based on

the join here.

As shown

in the lower half

of the pipeline,

instead of directly computing

the feature ID

to sample's data set,

we actually generate

two data sets here.

First is the data set

that maps partition ID

to sample list.

This can be done

using the map partitions

with index transformation

in Spark.

Another data set

is a mapping from feature ID

to a list of partition IDs.

Constructed through

the inverted index approach.

Notably,

rather than having

redundant data samples,

now we only have

redundant partition IDs,

which is much cheaper.

Next,

through the two cheaper joins

here,

we can achieve

the data set

that we want

to be able to calculate

gradient descent locally.

So,

for example,

we can map

partition IDs

in each partition

or locally,

we mean,

like locally

in the executors

when possible,

or when candidate

RDD

might look like.

It's like

a mapping

from partition ID

to a tuple

containing

a waitlist

and a sample list.

And it's your job

to think about

and figure out

how to use

join operations

to achieve

this task.

Think about

how you can

use

these two RDDs,

the samples partition RDD

and the feature partition RDD,

together with

other existing RDDs

like weights RDD,

to derive

the weight sample RDD.

Brainstorm yourself

and think

about

how you can use

the inverted index

join operations

to join keys

and eventually

get this step.

Furthermore,

as for

RDD partitioning

and join optimization,

we have

some suggestions here.

First,

we would like

to introduce

some RDD partition operations.

So,

for

partition level

map

and reduce

task,

you might want

to take a look

at map partitions

and partition with index

and also

glom operations.

And for

repartition operations,

you might want

to take a look

at map partitions

with partition

and partition by

is useful.

Second point

is that

before doing

join operation,

if you do

a partition

by key operation first,

then the

join operation

might be cheaper.

One interesting

thinking question

is that

can you partition

the data set once

and do not disturb

the data set

and you reuse

the data set

for multiple joins

so that it saves

you the

shuffle cost.

The third point

is that

you might need

to

think about

the number

of relations,

its relationship

with the number

of features

and number

of cores

so that

each record

can fit

in memory

and the

join operation

can have

a good

performance.

One trade-off

here is that

if we are

having higher

number

of partitions,

then the

join operation

would be slower

but it will cause

less memory.

However,

if we have

less partition

numbers,

the join operation

would be quicker

but it will cause

a lot of memory

and even

cause

out-of-memory

errors.

So,

explore

and evaluate

the impact

yourself

and figure out

some general

solution

to decide

the number

of features

and number

of cores.

And please pay

attention that

we

do not

allow hard

code

and make

separate

submissions

for each

test cases.

For more

information,

please refer

to the

write-up.

In addition,

you might

always

want

to

use

the same

number

of partitions

in your

test cases.

In this

example,

there are

at least

two possible

cases that

the output

size might

blow out.

First is

the output

of the partition

of the join

operation.

Another is

the output

of reduce

by key

operation.

In both

cases,

you might

want to

try using

repartition

or partition

by operation

to maintain

the

result.

In this

example,

you might

be able

to find

situations

where

RDDs

can be

used

across

iterations.

For example,

the samples

RDD

and the

features

indices

are

read-only

data

without

any

updates.

So you

can

consider

most of

the

important

hints

for P2-3.

I'd

like to

give a

brief

introduction

to RDD

join

operators.

From a

logical

view,

the use

of RDD

join

operation

is for

combining

two key

value

datasets

together

so that

pairs

with

RDDs

can

be

combined

locally.

On the

other hand,

from an

implementation

perspective,

the goal

of the

Spark RDD

join

operator

is to

bring

corresponding

keys

from

each

RDD

to

the

same

partition

so that

they

can

be

integrated

into

the

RDD

join

operation.

For example,

if RDD-A

and

RDD-B

don't

have a

partitioner,

we'll

have

two

shuffle

operations

here.

However,

if we

specify

the

partitioner

before

doing

join

operation,

we

can

use

the

hash

function,

which

will

use

the

default

hash

partition

algorithm

to

distribute

key

value

pairs.

Another way

to specify

the

partitioner

is

using

partition

bind

operations,

in

which

only

one

shuffle

operation

is

required.

In addition,

if both

RDDs

use

the

same

partitioner,

which

means

firstly,

they'll

have

the

same

number

of

partitions,

and

secondly,

they'll

use

the

same number

of

partition

bind

functions,

which

saves

you

a lot

of

shuffle

cost.

And

also,

some

other

general

optimizations

for

RDD

join

operator

are as

follows.

Note that

you

might

not

need

to

do

a

join

operation

to

reduce

the

result

size

for

the

join

operation.

Secondly,

pre-partitioning

with

non-partitioner

can

reduce

the

cost

from

shuffle

operations.

Besides,

a common

strategy

is

to

test

one

iteration

at

a

time

or

two

iterations

to

have

an

estimate

about

what's

your

runtime

for

the

whole

test

cases.

Subsequent

iterations

are

about

20%

cheaper.

It's

especially

useful

for

test

cases.

And always,

please

start

early.

Yeah.

That's

all for

the

hints

video.

Hope you

enjoyed

this project

and best

wishes.

Thank you.