Excellent. Thank you very much. So now we can start with the talk. Rafael is going to talk
about effortless distributed computing and Python. So take it away.
Can you hear me well? Fine. Okay. So my name is Rafael. I'm a software engineer and
for the past few months I contributed to a few new libraries for client. And I would like
to showcase them. So all the libraries are related to distributed computing.
So I would present the libraries the first one. It's a we named it scalar. So it's a
kind of distributed system for Python. So you can send Python task and it will
put them to some workers, different computers, do the computation and return to result.
On top of that we made a profit. So it's a very small library that you can use to
do map produce. So I will explain what map you use is. And the last one is
paragraph. It's another library that we developed to do distributed graph
computation. So graph is when you have dependencies between task. So we start
with scalar. This is really the core of what we did. It looks a bit like task for
those we know, you know. So since Python 3.2 you can use the process
put executor to submit task to different processes, Python processes on your
machine to do some kind of parallel computing. So this is an example. So you start a
four workers in this case. I'm not going to use all of them. And you submit two
functions to square root with different parameters. And the executor will not
return the result immediately with a future. So a future is just an object
that will add some point content the result. And then you can just print
and use the result. But Python will block if the result is not ready yet. So
if the computation that happened on another process didn't finish yet. So
it's an easy way to do a parallelization in Python. And what we wanted to do is
to not only be able to send processes to different cores, but also to different
machines. So we created an infrastructure. So instead of using the process
tool, you use our client that connects to a cluster. And you can submit task
the same way. So it's a total transparent. It will serialize the function,
the argument, and then fetch back the result. It's similar to task. I can
talk about the difference after the talk if some are interested. So the
architecture is like this. You've got the scheduler, which is the central point
of our system. And then you can have as many workers as you want that are
on different machines. And the scheduler is responsible for accepting functions
and tasks from the client and redirecting them to the worker that had
resources available. So we'd like to show a small example. So we are going
to make a parallel implementation of the Monte Carlo approximation of Python.
So the approximation of Python using Monte Carlo is basically you take a
square and you put random points. You generate random points on that square.
And you count the points that fill that four inside the circle that is in
this square. And on the example on the left, if you count the points that
they are in the green inside the circle, you divide that by the ratio
of points. You multiply that by four. You will get an approximation of
five. So three to two is pretty close. Points you had the closer to five
you get. So this is the example I'm going to use in the next slide. So if you
want to do that in Python, first you have to define what is when a point is
inside the circle. So you take the trigger metric equation for the circle.
You check the point is inside the circle. It doesn't matter much. It's just
for example. And then you do the Monte Carlo estimation. So you generate
random coordinates. So x is and y's. And you filter to the points that are
inside the circle. And you count them. You multiply that by four and divide it
by the number of points. And you get an estimate of five. So the more points you had.
So this is the function run. The more points you had, the closer to five you
get. And this is a sequential version. And if you run it with one hundred
thousand points, it's really quite slow. Maybe a few seconds. But it gets closer to five.
So what if we would like to scale that further. And first do it on multiple
course and then on multiple machines. So we're going to make another function
that we'll use the first one. And the function what it will do is that it will
split the number of points that you want to generate. By it is case one hundred.
And for each of this subset of points, it will start the task. So this is here,
the submit. It will start 100 tasks that each have one hundred of the number of
points. Then you average the result. So this sum is during the average. So all this is
compared to the sequential one. So the sequential one, if you want to do it on one
billion points, it will take about seven minutes. So it's quite slow. If you use the
byte and built in executor with eight course. So this is this laptop. You will get it in
five or five minutes. So it will take about fifty seven seconds. So it's already way faster.
And if you use all scalar client. This is on a big cluster. We have one
thirty five course on that one. You see that the computation is twelve
seconds. So it's thirty five times faster than the other one. So this is
basically all you can do prior computation with scalar. We have additional
one. This one is equivalent to edge top. So you can see those status of the server.
All the course, you know, how much CPU is it to have. You know, how many
clients are connected to the cluster. And you also have the status of the
skidder, which is read the sunflower point of or infer. We have
feeder recovery, which means that if a work dies, why it is computing a task,
the skidder will immediately reboot all the tasks that the worker was
responsible for to available workers. We have also dynamic load
balancing. Like if one worker has too many tasks and just cannot
make progress that might happen if some of your tasks are way longer to compute
than others. It will reboot all the tasks that the worker
accumulated on available workers. So that's for skidder. And on top of that
we made a library that you can use to do map reduce. So I'm going to
explain what sorry map reduce is. So another example, you want to count
the words that you have inside the text. So you get this count words
function. So it accepts a list of lines. So it's every sentence is
an item in the list. And you just go through all the lines and all the
words and you just accumulate and count the words. And then when you
run it on a small file, you will get obviously that the most common words
are d and then off. But what if we want to do this on a very large
data set? Like something that is gigabytes. And what if we want the result
to be fast? You can use the map reduce pattern. So map reduce is
basically you take the inputs. You split the inputs in different batches
or partitions. You process all these partitions and batches
separately and then you have a function that combines the result. So
paraphernal just helps you doing that. So it's a decorator. So you
take the original function. You don't change anything to the function that
you have. You just say that okay, I've got this function. I want to
paralyze it. I want to check the data by batches. So in this case,
we have predefined function for list. So the first argument of the
decorator says I want the input data to be split it in whatever
the size is, but in batches of lines. And then I want the
executive function and finally I will combine the result with the
sum. So you can sum contours and you will get the little
option. So by doing this, you can call the function to take
transparently like this one with a way larger file. And actually
it will run on multiple computers if you're using scalar
file. And something that is very important with
Parthen is that you don't specify the size of the batches.
You don't specify all you're going to split the, I mean,
you're not going to specify the size of the small task. So you
don't know how many tasks will be created. Because we,
finding the optimal batch size for computation is quite hard.
Like, if it is too small, so if you create too many tasks,
you will spend a lot of time just trying to transfer
data communicating and doing some synchronization stuff,
maybe IPC, you're running on single machine. And if you
sub-task or too large, like if you split the data in two
large chunks, you will not get that much
Parization. So you want something that is smart about that.
So we use machine learning. So the decorator won't use it.
This is the debugging information. It will automatically
automatically deduce what is the optimal size of the
partitions of the batch. So in this case, it deduced
some tries that computing the function in batches of
1,660,600 lines. It's actually quite efficient.
By doing that, you will only spend 5% of the time doing
communication and IPCs. And you get a nice speed up.
So this is the third transparent. And that's probably the main feature of
our map with use thing.
Lastly, we have paragraph. So it's, again, another library
that you can use on top of scatter. And this one is really
designed to do graph computation. I've got a third example.
So you want to generate a report. And your report, it opens
a file. It reads the file. It pre-process the files.
But it also goes to a database. And it will extract
the email, the email addresses. And then finally, it will send
the report to all these email addresses. So if you are into
distributed computing, you will notice that actually you can
run this to a set of code, a book of code,
super key, and in a personalized way. Because they're not
dependent. They're only dependent here. So what we have is
two additional decorators that allows you to define graphs in a
declarative way in Python. So first we have the
delayed one, the delay one, you will define what is your
graph leaves. So what the smallest function that you don't
want to utilize. And we just had the delay to all the
sub-task. So remember, we had this read data file, read
progress table, extract emails, all these functions, all
the ones that are in green and orange. You just said that
this is a release of your graph. And then the parent
function, the one that is coding the other function, you
said that it is a graph. And by doing that, when you
request the graph representation of this
computation, it will be actually not really run the
function, so it will inspect the function. And it will be
able to generate a graph that will present the
computation. And here we see that we have two paths that we can
take to do the computation in a distributed way. And once
you have this graph, you can export it and run it on
scale. So we have a special function. You can also
run it on that. It is the same format. And that is
basically the ID. I am a little bit
early. So I have a lot of time for questions.
Thank you.
Thank you so much.
Just before the question, we released that a few
months ago on the Apache 2. And we welcome
contributors and issues and if you have just
it is on GitHub.
Thanks for the presentation. For the
graph representation, can you represent more
advanced graph structures including what? Can you
represent more complicated graph structures?
You are outta
fail Ju.
We had a couple of questions and questions.
In comparison to the concepts, in fact, various simple
structures and equations. So this one is a very simple one.
what how many levels he wants, what the graph library is doing is that it's
it's overriding all the Python operators. So when you run the function it actually
doesn't run the function, it's just generate a kind of syntax tree and
we extract the graph from that. So unless the operator is not correctly
designed it should work and the format for which we generate the graph is the same
as desk which is part of the library so you can run it on desk and on
skater, what was it the second one, what is it the second one,
the format it's pretty standard, it's a Python dictionary, I'm just going to read
one question that was written in the chat so it says with full scalar, how does
it include, how does it handle Python dependencies say if you need a
pipeline package installed on your for your task do you need to make sure
that it's on your notes in advance? Yeah so we are using CloudPycle so it's
basically the same constraints as CloudPycles so you you might want to have the
libraries installed, however you can override the serializer so if you want to
do something more complex you could technically do it but yeah that's one of the
complexities you have to to install the libraries on all the machines
daily, hello thank you for your for the talk, what is the relation between
scalar and something like salary? I never used it but it's mostly a queue, I don't
think it it uses the executor interface but there might be a wrong I never used it
yeah sorry I don't know much about sorry so I can I can't read
answer hi yeah thanks for the interesting quotation I was wondering about
testing do you have a support like like if I had these decorators can I
disable them you're in testing or how do I yeah yeah for the graph actually so if
you don't call the two graph it just looks like a Python function so you can
just run it sequentially and for powerful we actually renamed the original
function so you just had two underscore sequential and you get access to the original
function you can also disable it on the system wide system by coding some
function and the second question if I may with par from there's no explicit
connection to to scalar is it like something under it? Yeah it's okay you have to set
up this in the you can either create a context you know the width something
in Python or you set up the backends on the on the world process that's the
two way we we under that so you can define the function and use different
backends and call the function using different backends at the same time
yeah thanks for the presentation I have two questions one I'll do you
set up the cluster I think you mentioned cloud python is that now it's just
we give our processes so we just have we get back to do so there's two things
are just Python programs so you just start the programs and they all connect to
some IP address or whatever you want your configuration it's just Python
commands so all the configuration is in the parameters so in the arguments
okay thanks and on the fillers scenario you you saw out to recover from a
worker failing what if this scheduler is that is yeah you don't yeah we
we we I have a question in terms of the worst case scenario do you mind going
to a slide where like one of the cluster nodes dies yeah so I'm wondering
just because of the syntax I don't this it takes into the pi example like you
run the scheduler runs all the jobs and then at the end it's some tip together
so you can get like the result of pi and a worst case scenario let's say the
cluster dies or all the nodes die like one of the jobs never completes like
what happens to that computation what happens to that syntax where you're
adding stuff and you end up adding like a none let's say or you mean if you're
process so you're running a process they're so long that let's say like
one of the nodes just dies and you never get a response and you're adding
over multiple processes like does it return a none and then it tries to add a
none and then you get like an air Python I'm not sure I understand but the
workers are constantly connecting to the scheduler so if she lose the
connection it's a kind of thing so if the latencies became too big like one
minute the worker is we'll be considered as dead all the task will be
redirected yeah but if all the if the workers never able to answer what does
the scheduler return to that like if it never gets an answer for a job oh but if
this one dies the scheduler keeps the task on it yeah it's
Department we have a parameter for that so if you don't store the task on the
scheduler obviously it will answer an exception to the client but by default we
store the task object so the arguments and the function code
basically and we can reboot the task thank you I had another question for the
graph execution one can the graph be modified why it's running or you have to
compile it beforehand and then it could technically modify it yourself like
as I said it built it built a kind of abstract tree of the computation as a
Python dictionary so you could technically access it and modify it but the library
doesn't do it how do the how does the system handle data exchange between
workers like a result of a function is being in the graph is returned and is
needed by another function I mean for the graph computation yeah yeah so it will not
send the task unless these are needed so we have it's ended on the scheduler
side on on the client side so we're not queueing the task to the workers until we
have the argument for the client weights for the data to come back and it's
like okay now next task that schedules will get that input I'm not sure
I understand well if you have like a function and it there are dependencies
between them right data dependencies so one function as an input take the output of
another function yeah like how does that data flow from one oh and it might be
scheduled and it doesn't get back to the client unless you request it okay so
it's the scheduler it will so let's say that the work on top wants to execute
a function that executed there so we get the data back to the scheduler and then
it gets back to the the other worker that needs to input just a quick one
maybe I got distracted and haven't got that part the batch size that you
mentioned is it's something that you need to make the book keeping off the size
of it or is it somewhere automatically in the background calculated it's calculated
so it's calculated yeah it'll give any eye no it's basically machine running
it's a regression so it looks at the function when you execute it and it tries
different batch size and after a few tries it will deduce that the optimal
one is whatever it is and it will continue to learn so if you change the parameters
for example and suddenly the function doesn't behave the same it
starts to be learned okay thank you so I think that's it in terms of
questions thank you very much ah that's one more sorry I've got 58 seconds this is
six one seven so coming coming from communities it looks similar design the big
problem communities are CPU limit memory limit even I owe limit so how
do you deal with that so we've got a features I'm working on it it will be
released next week but we have what we call tags so if we work on hand can have
specific tags so it's just symbols strings and you can say OT1 as a lot of memory
and and then when you submit the task it will only be redirected to the
work your that F this specific tag so yeah but my kind of handle that and if you
want to limit the CPU usage inside the process you have to run in C group so
things like that you have to deal with that on the operating system level okay so thank you very
thank you very much