Latest / Agentic Conversations (formally mlops.community) / Real-time Feature Generation at Lyft // Rakesh Kumar // #334

Transcript

Real-time Feature Generation at Lyft // Rakesh Kumar // #334 · Agentic Conversations (formally mlops.community)

0:00
Some of them are like precursor as well.
0:02
So when we initially introduced the pipeline, right, we were
0:07
able to cut down almost like 60% of the code base.
0:11
Wow. So I'm Rakesh and senior staff
0:19
engineer at Lyft and I like Latte.
0:24
We recently had your colleague Josh on here and he was talking
0:29
a lot about the data science side, the algorithms, the
0:34
modelling side of things. Now you're coming from the
0:37
engineering side and we get to talk to you about real time
0:42
machine learning, which is everyone's favorite topic I
0:46
think because it is so hard to get right and you all have been
0:52
getting it right for a while. Maybe we can break down first.
0:55
What do you do in real time machine learning use cases with
1:00
at Lift for? So, real time use case.
1:04
A real time machine learning features mention use cases.
1:06
They cover most of the critical business use cases.
1:09
And that's a mouthful. So I will bake it down.
1:13
So what we do is we try to process real time data that is
1:20
coming from the users, devices, various services.
1:24
We get those data, we process that data and we transform the
1:28
data, we aggregate the data and then we pass it down to our
1:33
machine learning model. And these machine learning
1:36
models are basically powering some of the use cases.
1:40
One good example which was covered earlier in in the
1:43
podcast that was real time forecasting.
1:46
So for any region or any city, any venue, we want to forecast,
1:52
you know what will be the demand, what will be the supply
1:55
at any specific time. So that is 1 use case.
1:57
The other use case is search prices.
2:00
That means we want, we generally monitor marketplace for any city
2:06
and if there is any kind of imbalance in demand and supply,
2:10
we want to accordingly adjust the prices.
2:13
For example, if there's less number of driver in that area,
2:17
we want to increase the prices so that like we slightly dampen
2:22
the demand. Yeah.
2:23
So that way like we are trying to adjust all those things.
2:26
So these are two examples, but there are a couple of other
2:29
examples as well. Yeah, and the hard part about
2:33
this is the real time list to it, right?
2:36
That adds a bunch of extra complexity if it's that online
2:42
feature generation and then creating that model that can be
2:47
making these predictions in real time.
2:49
Right. So to just cover the full
2:52
context in feature generation part, there are primary 21 is
2:57
the offline generation and the other one is real time
2:59
generation. In offline generation, you have
3:02
the data, in offline data store you have time to you know
3:06
process the data and generate the feature.
3:08
You do not have that tight latency requirement, right?
3:12
But in the real time, what happens is that like you need to
3:16
process millions of data points in real time.
3:19
That means it could be second to sub second latency requirement,
3:24
right. So as you are getting the data,
3:26
you are processing all these things and you have to provide
3:29
that data as soon as possible to do all these real time machine
3:33
learning model so they can make all these predictions.
3:36
And it is very important for a Lyft to process all these behind
3:42
real time. You do not want to serve still
3:44
data to model because then your model will be drifting apart
3:48
from the actual actually what is happening in in, in the world,
3:52
right? Let's say you your model is like
3:56
or your pipeline is slow by 30 minutes.
3:59
It may not be accurately predicting the current market
4:01
condition. You will be giving prices which
4:05
are stale, which is old and whatever you're trying to do, it
4:09
may not reflect correctly. So it's really important for us
4:13
to basically process all these data really fast and just
4:17
provide it to all these machine learning models.
4:20
Yeah, you can potentially lose a lot of money for the business.
4:24
And we had, we had seen that as well like in terms of latency,
4:31
if we degrade our latency requirement or we take a longer
4:36
time, it affects the performance of the model.
4:40
Yeah. So it is very important we try
4:43
to minimize the latency as much as possible and it could go to a
4:48
second or sub seconds as well. Well, I want to talk about some
4:53
tricks that you've done to make sure that latency is as low as
4:58
possible. If I am understanding it
5:02
correctly, when you say real time, you're one of the few
5:05
people that mean real real real time.
5:08
It is not. It's like a 5 minute window real
5:12
time. It's more half a second type of
5:15
real time. So I, I think that you had a
5:18
kind of a really good question. What is real time, right?
5:21
Different people have a different definition about the
5:24
real time. Sometimes people say synchronous
5:27
call is real time, some people say you know asynchronous
5:30
processing but in real time that is it real time.
5:33
So for this talk we are talking about the async processing like
5:39
we are not talking about scenarios but for this specific
5:43
talk we will be talking mainly about the async processing.
5:46
So and there also there are a couple of what do you call it
5:51
use cases. You mentioned about 5 minute
5:55
window, one minute window we do do that as well.
5:59
It completely depends on the the use case, right.
6:04
For forecasting use case we have aggregate like we aggregate data
6:08
for one minute window and five minute window as well.
6:11
And then we providing to them. In some cases we also have a
6:16
short term historical data and then we provide it to them.
6:19
So it completely depends on the use case.
6:22
Most of our use cases are based on one minute aggregated data.
6:27
That's how our models are designed and that's how or
6:31
that's our that's how the pipeline are supposed to provide
6:34
that data to to the models. So it is not the extreme side
6:40
where like it is you know sub second or so or like when you're
6:44
when you're talking about a synchronous call, it's more
6:46
about the asynchronous processing of the data.
6:50
Is it like the complexity and the difficulty increases
6:57
exponentially if you're going from that 5 minute real time to
7:03
one minute real time? Or is it something where you can
7:08
create a system and once it's in place, you can get to a certain
7:14
point where you've got that cushion and whether it's 10
7:19
minutes or it's one minute, you're good, but you start
7:22
hitting the ceiling. If you're starting to talk about
7:24
the sub second latency, like I'm trying to gauge how different it
7:30
is to be at 5 minute real time versus one minute real time
7:34
versus millisecond real time. So if you go further like if
7:40
you're trying to look at the sub second latency and you're
7:44
comparing with the synchronous processing, it is not the case
7:47
there. It is like most of the cases
7:51
asynchronous call is really with one's request.
7:55
But our processing is based on the aggregated data.
7:58
That means we are getting all these data from different
8:00
sources and we are trying to aggregate them based on the
8:04
geohashes. So geohashes is a size of a city
8:07
block or so. So like and you can imagine like
8:11
we are getting this data from different cities across
8:15
different nations, you know, and then we are trying to process
8:19
that data and then we are processing the data and then we
8:24
have to go to the city level block and aggregate the data.
8:29
Coming back to your question, like where is the nice cushion?
8:34
I would say it depends, it depends on the use case and the
8:38
data volume that we are trying to process it.
8:41
I would say if it is smaller window, it's much easier to
8:46
aggravate because you are accumulating less data and you
8:50
can you know, emit the data faster.
8:53
But if you have a longer window, that means you are connecting
8:58
that data in memory for a longer time and then you're emitting
9:02
that data. So you're processing and what do
9:06
you call the memory utilization increase significantly.
9:10
And we had seen use cases where we tried to do 30 minutes
9:14
aggregation. It was working, but it was not
9:18
scalable. And yeah.
9:20
Got expensive? Yeah.
9:22
So one minute or five minute, those are pretty good and good
9:27
use cases. Now, can you walk me through a
9:31
bit of the evolution of how you've tackled this specific
9:38
problem, the real time problem while being at Lyft?
9:42
Because I feel like you've been through different iterations.
9:46
I will divide this value evolution in two or three
9:50
phases. First phase, like when we were
9:54
very early phase, when we were not even using all these
9:58
streaming engines to process all these data, what we had at that
10:02
time was Cron job based processing.
10:06
So it will get up at the top of the minute.
10:09
It will try to get all these data from the Kinesis processes,
10:14
store it in a kind of temporary location.
10:17
Most likely that is then the next Cron job will get up.
10:20
It will, it will process that data and that's put it and
10:24
imagine every stamp is doing this processing the top of the
10:28
minute. So even though the very first
10:31
process, very first Cron job has finished, it starts within one
10:35
or two seconds. The next one is not going to
10:37
start immediately. It will take a lot longer.
10:40
It it was working but like as we are growing it was not a kind of
10:45
a scalable solution. Our models were not also
10:48
performing well because inherently the system has
10:52
latency. So what we try to do is like OK,
10:55
since our model will not perform better, like if there is
10:58
inherent latency, we want to reduce it.
11:00
So we research on like what are technologies are, how we can re
11:04
architect the solution and we after our research, we realized
11:10
that we can use some streaming based solutions which can
11:13
process this data. And there you do not need to
11:16
wait it wait for the next step to get triggered at specific
11:20
time in the stream processing. As soon as the very first
11:24
operator has processed the data, it will send it to the next
11:27
operator or next operator downstream and it will process
11:31
it. So it's more about like when the
11:33
data is available, it will process, it does not have to
11:35
wait there. So initially.
11:39
In this you're talking about like Flink, right?
11:41
Right, so and we use Apache Beam and Apache Beam is using a Flink
11:48
underneath. We initially did Apoc and
11:54
initial results were great, like it worked fine.
11:57
We also choose Apache Beam Python as the game, mainly
12:00
because we are Python shop. Most of our services are written
12:03
in Python. People are comfortable in Python
12:06
And you know, our models are written in Python.
12:08
So we'll start. OK, maybe we'll just use it.
12:10
The interesting thing is that though Apache Beam is in like we
12:17
we are using Apache Beam Python SDK, but it is doing most of the
12:21
processing in the Apache Flink and Flink is JBM based
12:26
technology. Yeah.
12:28
So it, it was a good, you know, hybrid solution and we used it
12:32
for what we call real time search pricing or or prime time.
12:40
And it was, it was good when we launched it for one or two
12:44
reason, regions or cities. But when we started scaling it
12:48
out, then we realized that, OK, this is not going to scale
12:53
because the way we were processing it, our data, it was
12:57
not very performant, mainly because we were doing most of
13:02
the processing in one operator. That was 1 issue that was easy
13:07
to fix it. You divide the operator, it is
13:10
perfect, right? It was right.
13:13
Then we hit the other problem where you know what happened is
13:18
that even though we separated out the processes, it was
13:22
processing it. Then we realized that some of
13:25
the nodes are like, you know, 90% CPU relation.
13:29
Some of them were not even doing anything.
13:31
It was like 10%. And the problem is how we, like
13:35
most of our cities are, I would say the traffic is divided based
13:40
on the cities. That's how our models process
13:42
all these things, right? And what we were doing is that
13:46
we were routing all the traffic from a specific city to one
13:51
particular Shard or one particular node.
13:54
And some of our top cities are of course right in New York, San
13:59
Francisco and all those things. They have almost 80% of the
14:03
traffic. So now imagine that 80% is going
14:06
to one node and the rest of the other are going to other nodes.
14:09
That was a really classic, you know, heart shot issue.
14:13
And we look, oh, OK. And that's interesting.
14:16
The distribution here is not adding up, yeah, so.
14:22
Then we realized that we we are not using it perfectly.
14:25
What we need to do is we need to divide that further processing
14:30
you. What you do is instead of city,
14:33
you divide in terms of Geo hashes.
14:35
Geo hashes again, there's a city block.
14:39
So what we do is as soon as we get the data, we say, OK, what
14:43
is the Geo hash for this particular event?
14:45
And then we started based on that one.
14:47
So from, you know, from you know 300 plus series then it further
14:54
translated into almost millions of Geo hashes.
14:58
So and these Geo hashes are almost uniformly distributed
15:02
across different nodes. So in a way like we, we are not
15:05
running into any kind of hot chart issue that the load is
15:08
uniformly distributed across different nodes.
15:12
And once most of the data processing is done, most of the
15:16
heavy computation is done then and this how you competition is
15:20
like, you know, filtering those data, aggregating it and any way
15:23
we wanted all these aggregated data based on the geohash then.
15:27
And once that is processed, you have one number and then the
15:31
geohash. So it comes down to a very small
15:34
data set that you can again recharted to region and that's
15:39
how like one region can get all the relevant geohashes.
15:43
So effectively it's small data and that does not run into any
15:47
kind of hard shooting issue. Those bottlenecks aren't there
15:53
now. I feel like I've talked to
15:57
somebody at Lyft before. It wasn't on the podcast
15:59
specifically, but they were in the community and they were
16:03
mentioning how having to create different configs for each of
16:11
the setups that they had ended up driving them nuts.
16:16
And I think they ended up going and creating a company around
16:19
it. But as you're talking about this
16:22
and you're saying, hey, we've got all of these different
16:25
models they're pulling in, or we've got all these different
16:27
nodes, they're pulling in the geohashes.
16:30
And then you also have the different clouds that you're on
16:33
and you have the different regions that you're playing with
16:37
there. Is that something that you
16:40
looked at and that was affecting you?
16:43
Also when it comes to having to figure out the config on this
16:50
model or this node or this Geohash, I'm not sure how you,
16:55
how you broke it down is with this region and it's got the,
17:00
these things that we need to be aware of because are you running
17:04
just thousands of models across the world or hundreds of
17:09
thousands? I I don't know what what that
17:11
looks like and how that works. So most of the real time
17:16
processing in that case we use model per region.
17:22
And so that is like one simply simplistic simple view of the
17:27
world. But that could be also further
17:31
divided into sub regions like and there's a couple of use
17:36
cases there. So and like if you, let's say
17:41
San Francisco, it's a big city like and sometime if you're
17:46
trying to process the data and trying to give it to 1 model, it
17:48
may not be performed accurately right in different parameters.
17:54
So what do you want to do is like you want to divide that
17:56
region into sub regions. So that way you independently
18:00
process all these regions. There are trade-offs there, but
18:03
at least on like if you are breaking down to a smaller
18:07
region, it's much easier to manage them.
18:10
So that's how like we used to do.
18:12
So earlier we were processing the region level.
18:15
Now in some of the use cases we have broken down into sub region
18:19
level and that's how we process it.
18:21
But on the pipeline side we are still doing on Geo hashes.
18:26
We are processing Geo hashes and it it does not matter like how
18:32
you consume it, we can always transform that data either to
18:35
sub region level or region level.
18:37
Of course, like the our customers, they define how they
18:41
want to consume the data and it's quite easy to just
18:44
transform the data and then provided.
18:46
And in a way it's much better for us that we further divide
18:50
that region into smaller chunks. So that way we do not run into a
18:55
hard shot issues as well. I like how you mentioned there
18:58
that our customers, I'm assuming that's like Josh who I spoke
19:03
with a few weeks ago, that's the data scientist.
19:06
They get to define how they want that data, how often they want
19:10
that data, all of that type of thing.
19:12
And I think the fancy word for that these days is a data
19:16
contract. You all have put into place a
19:19
data contract. You have the producer of the
19:21
data and then the consumer of the data.
19:23
And you're able to stay accountable so that if you're
19:29
changing anything with that data, Josh is in the loop.
19:33
Or if Josh needs anything changed, then you're in the
19:36
loop. And I wonder how you got to that
19:41
point. So how we got it to that point?
19:46
So what we generally do, there are two strategy that we use.
19:51
In one case it is like off the shelf, you can use that pipeline
19:55
generate feature by by yourself and most of the cases they are
20:01
very simple aggregation. But in case where you have very
20:06
specialized need and some complicated business logic, in
20:11
that case we hold their hand and say that you provide me the
20:15
contract, I will build the future for you.
20:19
And we like they generally what happens they, they create the
20:24
data from the offline data stores.
20:27
They create, come up with a query, they generate all these
20:30
features in offline mode. That's their model.
20:32
If everything is looking good, they say, OK, the performance is
20:35
really looking good, I want to productionize it, I want to make
20:39
it real time. That's when we come into play.
20:42
And then say we convert that query into a pipeline
20:46
implementation. So like we transform that query
20:50
into a pipeline and our contract is generally that whatever
20:55
features that you are able to compute using offline data, we
21:01
will be able to match that data with almost 99.9% OK.
21:07
And when we build this pipeline, we compare the real time feature
21:12
with the offline 1 and make sure that it is closed and and.
21:17
And this is also required because we have a tight SLA in
21:20
terms of, you know, latency as well as the quality and we do
21:26
not want to degrade on the quality.
21:28
So we have another framework or in house framework which
21:34
basically comparing the the real time feature and offline 1.
21:40
Offline in the sense that it is a source of crude our ground Rd.
21:45
we are comparing these two feature.
21:47
We are basically sampling the real time feature and then the
21:50
offline 1 ends and trying to compare like how different they
21:54
are if they are. That's cool.
21:56
Yeah, if it is, they are different than we.
22:00
There's alert define for that one and it will trigger and then
22:02
we we say why they are different.
22:04
We, we are but, but most of our features or I would say all of
22:10
the features that we generate on a real time feature platform, we
22:14
have that kind of observability in place.
22:16
So it's really the operationalizing of the feature
22:22
pipelines. Yes, that's right.
22:24
And so as a data scientist, they almost have it.
22:29
They've got it good. They can just develop these
22:32
feature pipelines offline and then throw you the shitstorm of
22:37
making them real time. Yeah, it is, It is interesting.
22:44
Yeah, like and then there, there are different consideration when
22:50
we are transforming that offline query into a real time query.
22:56
At the high level the results are same.
22:59
But when you implemented, you have to also consider like how
23:02
these pipelines are processing these data, how I can avoid the
23:06
hard charting issues. And like there's a lot goes into
23:12
while designing that pipeline. And some of our in house
23:17
engineers, they are explored in this and then as soon as they
23:20
get it, they query, they know like these are the parameters
23:22
which are not looking good. We have to do a further
23:25
processing. We have to further divide into
23:27
based on the number of Geo hashes and make it process in
23:32
such a way that it will not run into a kind of hard shaded show.
23:36
Yeah. Now the obvious next question
23:41
is, do you all have? Are you using some type of a
23:45
feature store? We do use a feature store and
23:50
most of the traditional feature store may not work for us,
23:54
mainly because they are point feature point feature.
23:57
That means you have a key, let's say I'm user, and then you have
24:00
a value. So let's say when this user was
24:04
onboarded, so you just provide a date, right?
24:07
Or whether what is the city where this user or requested a
24:14
ride something that this one's, it's very straightforward,
24:17
right? So that is 1 use case.
24:19
But the other use case where most of our models work, that is
24:25
aggregated, right? We try to figure out whether
24:29
there's an imbalance in the market and we do not look at the
24:33
individual session, we look at the holistic view of that city
24:37
or region. So we want to see how many
24:40
drivers are there, how many requests are coming from that
24:43
city, right? And in this case, like
24:46
reconsider that city or geohashes, right?
24:49
And we want to not get just the one point value.
24:53
It's like the whole aggregated value and it is hierarchical.
24:57
That means a city will have multiple G hashes and within the
25:01
multiple G hashes as well like we have GH4GH5GH6 so and all of
25:06
them are hierarchical. So regular feature store will
25:09
not work. So what we did is like we create
25:12
a geospatial based feature store there.
25:17
If you provide a region, then I can provide you all the
25:23
geohassets coming under that region and their aggregated
25:26
value. And you can choose the level at
25:30
which you want it like whether you wanted region level, whether
25:34
you want the GH5 level, GH4 level, you just provide that.
25:38
And then we can, we can just provide the data to you and, and
25:43
it is interesting and it works and, and it was able to scale
25:47
well as well. As I was saying earlier, some of
25:51
our model, they work at region level.
25:52
Some of them you know, at the sub region level.
25:56
Some of them they work at, you know, Geohash level, AGH 4
25:59
level. And the source of data is the
26:02
same, but they can, they can consume it differently, they can
26:08
ask for region level. We can still provide the region.
26:10
It is the same data. It's just that the accessing
26:12
mechanism is slightly different, but it has the same API.
26:16
And that has helped us a lot. Like you know we can generate
26:20
one feature and then this one feature can be shared by
26:23
multiple models as well. How are you making these
26:28
different layers of the geospatial data?
26:32
What is that even look like under the hood to be able to
26:38
serve up that type of offering on the?
26:41
API, yeah. So it's based on the
26:44
hierarchical. That means you have a region,
26:46
then you have GH4GH5GH6. So region might have a couple of
26:52
GH fours, GH fours may have a couple of GH fives and all.
26:57
And it's hierarchical. It's basically a kind of a tree
27:01
and when you're selecting anything like say you're
27:03
selecting a region, you know all the nodes under that tree or
27:08
leaves under on that. And then it's up to the the
27:12
builder or the data scientist to decide how granular they want to
27:16
get in this specific use case or for this specific model.
27:20
Yeah, that's right. And like how we store the data,
27:24
we generally store in two different ways.
27:26
One is the region level, like we store the entire thing and then
27:29
the rest of them we have divided into GH6 level and then we have
27:34
defined the mapping between GH6 and GH5.
27:38
So whenever someone sends us a request, we get the data from
27:43
the store, aggregate them for them and just pass it to them.
27:47
So it's it's basically when you look at the GH or Geohash the
27:55
the prefix are I would say postfix are always matching the
28:00
the latter part of the GSS is always matching.
28:04
So if I give you AGS 6 and if I tell you to come give me the
28:09
GH4, you're basically getting the let's substring of that one.
28:14
So you're basically just getting the four character from the GS
28:18
6. And that's how you're able to do
28:20
it. So it's it is very
28:22
straightforward calculation. Now the other thing was around
28:27
the offline and online piece, which I think it would be good
28:33
to go into how it looks offline when folks are doing offline
28:40
jobs versus online. And I, I think the online part
28:45
is quite clear to me right now, but I want to know what do you
28:51
do for offline? And also is offline just for
28:55
that ground truth? Is it for different use cases
28:59
that you're servicing? What is that?
29:03
So offline, so most of our real time data and they end up in
29:08
offline store for various reasons.
29:11
One of the primary reason is that you know like you want to
29:14
see historical data and when data scientist or MLA engineers
29:20
are working, they want to see how the data looks like.
29:23
They transform the data, they want to transform it and make it
29:27
as feature and test out with that model.
29:31
And this is what one example. The other example is like when
29:34
we want to validate the real time feature whether they are
29:37
accurate or not. Like that is another one as
29:40
well. Sometimes we can also do back
29:43
testing. That means if models misbehaved
29:45
in production, you want to go and replay and saying like why
29:49
they misbehave, you want to debug further.
29:51
So there are a couple of use cases there, offline and online.
29:56
I think offline world you get delayed data as well.
30:01
You have like, you know you can have all the data and you have
30:05
all the time to process it. But in real world or real time
30:09
processing, it's difficult. If the data is delayed, we
30:13
cannot wait forever for you to complete like and you do not
30:16
have that much of time. So what happens there?
30:19
Like you either just move fast, like you ignore that one and
30:23
move fast and you process it. So that's how it happens.
30:26
And earlier I was telling you that, you know, we have 99.9
30:30
percentile or percent matching data, but why it's still there
30:35
is a gap, why we are not able to match 100%?
30:39
That is the reasons in, in real time data may be missing.
30:43
It will not be delayed. Imagine person going through a
30:46
tunnel, somehow they don't have a network and then the device is
30:50
not able to send the data. And once they come to area where
30:53
they have a nice signal, nice connectivity, the data is synced
30:57
to the server. And that's how we get we get a
30:59
kind of delayed data. Yeah.
31:02
So one thing that I would love to get into though, is the the
31:14
actual decisions that are being made in these meetings when you
31:20
have to say, OK, we want to go and support XYZ, whatever XYZ.
31:27
Maybe, maybe it is a new technology, maybe it is a new
31:31
type of architecture or a new experiment where you're going to
31:36
change the system. How are you all thinking through
31:40
the trade-offs as you're thinking through these bigger
31:44
decisions? Because I can only imagine that
31:48
you're having your retros, you're also recognizing that
31:52
you're supporting quite a large scale and you want to
31:56
continuously be moving forward. You don't want to get stuck in
32:00
time, but because you have to make certain decisions and
32:06
create opinions, there are things that you're going to land
32:11
on that potentially can be a couple year relationship or a
32:18
decade long relationship with a certain technology or a certain
32:21
system, right? So there are so many
32:24
considerations go into this. Definitely one of the primary 1
32:32
is that whether this technology is a good fit for this problem
32:37
domain or not. It may be good fit right now,
32:41
but down the line it may not be good fit.
32:45
Things could change, traffic pattern could change, you know
32:49
there's a new technology which can solve this problem easily
32:53
where we have where we require to put in less effort.
32:58
As I was telling you about when we introduced streaming
33:01
processing, stream processing at the left earlier, it was based
33:05
on the Cron job. So Cron job may have solved the
33:07
problem when we were at small scale and it was OK.
33:11
But as we started growing scaling system then we realized
33:16
this is not able to scale and we need to look for the other
33:20
technology. So streaming was one good use
33:23
case where you don't need to write that much of code, you
33:27
need to write less code and it's still able to process the data
33:31
in real time. So that is 1.
33:33
The other and which is other problem which is also relevant
33:36
to this one is like how much effort that we need to put in to
33:40
build this technology or use this technology.
33:43
And that is most important one, as we try to move fast, it is
33:49
very important that the operational cost or development
33:52
cost is low on our side. So that way we can ship things
33:56
faster. The other part is the cost as
34:00
well. Over time we have you know move
34:04
from one solution to other mainly because initially when we
34:08
launched services it worked fine, but cost wise it was fine.
34:13
But as we are scaling up, we realize that certain places we
34:16
are like you know spending more, we could cut down those things.
34:20
Then we realize there OK, maybe there is other alternative or
34:24
maybe there is other way of like doing things and when we look
34:27
into it and we realize yes, there so cost is the other one
34:31
as well. We tried to look for what we
34:34
call technology which could be cheaper for us.
34:38
So that is the other one, the open source versus so open
34:43
source as well, if anything is open source, well adopted by the
34:47
community or well supported by the community, that is another
34:50
consideration. One more is the art ecosystem,
34:55
how well that technology or framework can integrate well
35:00
with our ecosystem. If it is really hard to
35:04
integrate it, then it may not be a kind of a good sell for us.
35:08
Example could be the stream processing.
35:10
We could have chosen Flink, you know, directly.
35:14
Yeah. But we did not use it mainly
35:17
because if we move from Python to Java, it would have been a
35:21
kind of very huge investment for us.
35:25
And and at that time, like we were scaling and we were also
35:29
trying new technology. You do not want to take too many
35:32
problems and try to solve them like you probably solve 1.
35:36
And that's why we said like, OK, we can choose a technology which
35:40
could be a good fit. And Python, most of the people
35:43
were familiar with the Python, so we chose a Batcha Beam
35:47
because a Batcha Beam supports a Python SDK as well.
35:50
So, yeah, these are the some of the considerations that we put
35:55
in before finalizing any technology or framework.
36:00
Can you imagine trying to sell a data scientist on using Java?
36:05
No, they will not like it it. Will not work at all.
36:10
Yeah, Python is well supported and I I think our customers are
36:14
data science and ML engineer. We do not want to introduce any
36:19
kind of solution, right, which will create unnecessary
36:22
friction. We want to create tools which
36:26
will make their life easier, right?
36:29
So and yeah, if it does not fit well in our ecosystem, even the
36:34
tool is great, it may not work well.
36:39
So and if there is any kind of bridge, there is a solution
36:42
which can bridge the gap, that would be great.
36:45
So we can probably consider both.
36:46
Then whatever the solution which can bridge the bridge the gap
36:50
and the final solution. Then it makes sense too that you
36:55
look at that bridge as how much developer time, how difficult is
37:03
this going to be to integrate into our system?
37:08
How often are you trying to look at what's on the market to see
37:15
if you can cut costs? Is this like a quarterly thing?
37:21
Is it a weekly thing? Is it a monthly or yearly thing?
37:25
I would say it depends on case by case.
37:27
I don't think we look at every day, but we do have monitoring
37:33
in place. If let's say if there's a
37:37
significant increase in cost, we know that why it has increased
37:41
and then probably if it has crossed certain thresholds, then
37:45
we will look into it like why they and if there is anything we
37:49
can do to bring down cost like without changing the technology
37:55
or framework, we would do it. Like if, if there is a kind of a
37:57
roadblock which is preventing us not or which does not allow us
38:03
to reduce the cost, then we have to think about, Roger, maybe we
38:07
should look for the other technology.
38:09
But it completely depends on the scenarios like how much you are
38:12
spending, what is the budget and what is our capacity to move to
38:17
a new technology. Yeah, We had bro hit on here
38:21
probably 3-4 weeks ago and he talked about how a huge boon has
38:30
been for folks that he works with recognizing if they don't
38:36
actually need the freshness of data that they think they need.
38:40
So just being able to checkpoint right data from going from one
38:45
second to 10 second checkpoints, he was like, you can save a lot
38:49
of money with that. Oh yeah, yeah.
38:52
So there's another process that we have is onboarding.
38:57
So whenever someone comes and say, you know, we want to build
39:00
a real time feature and like, and then our question is like,
39:02
do you really need a real time feature or do you need offline
39:05
feature? And we ask all those questions
39:08
based on their use cases, try to determine if they really need
39:11
it. If they do not need it, then we
39:14
point them to the offline data stores.
39:16
You can build there, you can add, because if you're moving
39:20
towards real time, it requires more effort, more money to have
39:26
that solution. If you do not need it, then you
39:29
are unnecessarily spending that much of time and effort to have
39:33
it. And that's why we said like we
39:36
vet all these requests and if they are genuinely or genuinely
39:42
needed. If that you know if it is it is
39:45
must to have it, then only we do it, otherwise we provide other
39:50
solutions, other offline solutions.
39:52
But why do you think that people are convinced they need some
39:57
type of real time feature? Why they're convinced that they
40:02
needed. So yeah, why did?
40:03
Why do they come to you thinking that they need real time?
40:06
Sometime, you know, they do not know that how much effort is
40:13
required, how much money we spend on, you know, generating
40:17
all these real time feature. You abstract all that away so
40:20
they think, oh just throw another real time feature in
40:23
the. So they feel it's, it is cheap
40:26
and and that's why they think that, OK, we can just have it
40:29
anyway it is available. So I don't we just use it right.
40:33
But as we own the platform, we generate all this, we know how
40:37
much effort and effort, time as well as money that we need to,
40:42
you know, spend on this. And we are really conscious like
40:46
when we made that decision. And so the data scientist get to
40:50
experiment with the offline stuff and say I'm a data
40:55
scientist and I discover that there's this feature that is
41:01
when someone logs on to lift, they search their ride, but they
41:07
don't actually get the ride and they log off of it.
41:11
We predict that X amount of X percent of times they're going
41:18
to come back and order that ride.
41:20
So I want that feature in my model something.
41:25
I know I'm oversimplifying the shit out of it, but just go with
41:28
me here on this. I want that feature on my model.
41:31
I see that offline. It actually gives my model some
41:35
help. Then I go to you and I say, hey,
41:41
I need this feature in my model. You come and you try and
41:45
convince me that I don't actually need it.
41:47
Isn't the model you say, Are you sure about that?
41:50
Can we maybe do something else? Is there potential for us to
41:53
just do this batch and eventually we settle on yes, I
41:59
need it. Then I go to saying, well, what
42:03
type of detail I want or I guess that would have already been
42:06
solved back in the beginning when I say what geohash do I
42:10
need? Yeah.
42:11
So like if I'm convinced that OK, this is a real time use case
42:16
and we are going to work on this one, then we will can.
42:18
And this is very super specialized use case where we
42:22
need more support to build it, then our ingenious will
42:25
transform that query into a pipeline.
42:29
And since these ingenious know how to build a pipeline which,
42:33
which is scalable, they will transform the whole processing.
42:37
So at the high level you will get the same data.
42:40
It's just that the processing could be slightly different.
42:43
And I, I think one of the optimization techniques that we
42:46
use and I have written in blog about it as well, we try to drop
42:53
unnecessary data early on in the pipeline.
42:57
And we trying to let's say like you are getting tons of data,
43:01
they may not be relevant to you. All right.
43:04
So as soon as we find this information, we drop it then
43:08
like we do not let the data pass to the entire pipeline.
43:12
And we have done that experiment like like early dropping versus
43:17
late dropping and we had seen at least 20% increase in efficiency
43:23
and even the latency improved. And that's when we made the
43:26
decision not like we will try to figure out all these data early
43:31
on and if they are not relevant, drop it and then process it.
43:35
And the the next step is also sometime the if you look at the
43:40
offline queries, they do mega joins each other like you know
43:46
tons of joins and all those things.
43:48
The real time processing does not work in that way.
43:51
You can do it, but it will very costly and unscalable.
43:55
What we do is even joins, we try to divide into a different
43:59
dimensions. So the high level idea is that
44:02
you divide the data into smaller groups and you perform the
44:07
operation. Smaller data is very manageable
44:11
than the bigger data. Yeah.
44:12
So it's more like a divide and conquer.
44:14
You are dividing bigger data set into smaller, process them and
44:18
then process it further. So that way you can manage it.
44:23
Otherwise it's really hard. If you had to guess, how many
44:27
pipelines are you up keeping right now?
44:33
Are we talking thousands, 10s of thousands, or hundreds?
44:38
So and definitely not 10s of thousands, we are really
44:41
conscious about it how we are building the pipeline.
44:44
There are two ways we can do it. 1 is like where people can build
44:48
a pipeline by themselves. The other one is super sliced
44:50
feature so super specialized like I would say 30 close 30-40
44:56
pipelines and these pipeline not necessarily generating only 40
45:00
feature like 1 pipeline itself can generate 10's and fifteenths
45:05
of feature. And when I say feature, it is
45:07
basically for one region and but they will be processing almost
45:12
millions of data per minute or so.
45:14
Wow. Yeah.
45:16
So that's how they're, I do not have the full account of like
45:20
how many pipelines that we currently run, but it definitely
45:24
somewhere in hundreds. Something that you mentioned
45:27
here is that there's self-serve in a way.
45:31
How does that work? Are the data scientists going
45:33
and they're trying to stand up MVP and then you come in a few
45:38
days later or however many weeks later and you make it
45:42
bulletproof? Yeah.
45:44
So I can sell services. Those are very simple use cases
45:48
like where you know processing only for smaller scale.
45:51
Like maybe you're processing individual user, you're not
45:54
doing a heavy processing of the data.
45:57
And those operations are really, really straightforward.
46:00
You know, like you're getting a user data and you're extracting
46:04
out what is the right request at time and then the storing
46:08
somewhere. Those could be the simple one
46:10
and since they are really easy, you don't need huge
46:15
understanding of how streaming processing works.
46:17
So they can just do it by themselves without much help.
46:20
And anyway, we have simplified the St.
46:22
KS and the whole infrastructure so they can bring up the whole
46:26
like the pipeline by themselves. So that is 1.
46:29
The other one is where it is super specialized, where it
46:32
requires very heavy lifting or heavy processing.
46:35
That's when some of the specialized team, they come in
46:39
and help help the data science person or ML engineer.
46:44
No. I know that people talk a bunch
46:47
about like the data getting skewed in offline, online type
46:53
of scenarios. What are you doing to mitigate
46:59
that? How do you go about that?
47:00
Is that in something that you had already talked about?
47:03
I think I did talk about that one as well.
47:06
And data skewness or heart Shard.
47:09
These are like interleaved things and whenever either you
47:14
are in offline online, I think online world is more affected by
47:19
that one. And we are like early on when we
47:23
were thinking that that will not affect us and we build the
47:28
pipeline and it was not super performant.
47:31
Like there were some of the nodes which we are getting way
47:34
more data and they had to do way more processing and the latency
47:39
was huge. One bad thing about stream
47:45
processing is that if anyone node is slow it will slow down
47:50
the entire processing as well. It it creates A barrier like if
47:54
one node has not processed it then the whole thing does not
47:59
move forward. It waits for that node to
48:02
process then the other node will also move forward.
48:06
So, so that is that was a problem.
48:09
And I, I think the data skewness, if you have to look at
48:12
the data like how the data is distributed and if you have a
48:17
good understanding about that one, then you have to come back
48:19
to the to the pipeline design and figure out how you can
48:23
further redistribute the data to avoid the data skewness.
48:28
In our case, region causes a huge data skewness.
48:33
It is like 8080% by 20% rule that means 20% of the cities,
48:40
they are generating almost 80% of data.
48:42
And so for those big cities we have to really cautious about
48:46
it. And in our case we just use the
48:49
geohash to further redistribute the data.
48:53
So from like you know, 300 plus cities to almost millions of
49:00
geohashes. So that way you are able to
49:02
redistribute the data and like you can avoid the data skewness
49:07
as well. I didn't realize that that was
49:12
the same thing. You were calling it hot Shard.
49:16
Yeah. Which I like that word now that
49:18
you're saying it. So data is, this is like there's
49:22
this concentration of data on one particular node or
49:25
something. And the problem what happens is
49:28
that like if there's a huge concentration on one node, then
49:32
that node is called heart, heart node, heart Shard because that
49:37
is the one which is processing most of it, but they are
49:39
deleted. Yeah, 'cause I guess I was
49:43
confusing it for the training serving SKU that you get with
49:48
the the the data sometimes and how you can mitigate that.
49:54
But it makes a lot of sense, this hot Shard and the data
49:58
skewedness, seeing the saturation on certain nodes
50:03
because there's so much more traffic on these different hot
50:07
cities. But then if you break it down to
50:11
smaller bits and you have these packages where you're just
50:15
looking at blocks a square block radius, then it's it's a
50:20
different problem that you're dealing with.
50:22
Exactly. And you you just redistributing
50:24
the whole data to all different nodes.
50:27
Yeah, it's a nice trick. I like that.
50:30
What other tricks do you got for me?
50:32
What else have you learned over the years?
50:35
Yeah. So I, I earlier mentioned that
50:37
we divided into two or three different phases.
50:39
So very first phase POC, we scale, we tried to scale, we ran
50:43
into some of the roadblocks, a heart shark one, one of them.
50:47
We addressed that and then we started creating multiple
50:50
pipelines. Then we realized that, you know,
50:54
it is kind of easy for to build a pipeline, but as you're
51:00
increasing number of pipelines, right, you're also increasing
51:03
the cost. And we closely look into all the
51:06
pipelines and we realize most of the pipelines are doing almost
51:11
very similar processing at the beginning of the pipeline or
51:15
there are a couple of operators which are doing something
51:18
similar. We thought like it can be
51:21
abstracted out and and centralized into one place.
51:26
So that way you do that processing only once and that
51:30
can allow you to save the cost. So what we did is we defined 2
51:35
stage pipeline. One stage is for preprocessing
51:38
all these data. So all these data filtering and
51:42
building a kind of state for the user.
51:45
So earlier each pipeline was generating one feature and and
51:50
if you have to generate another feature, you have to build
51:52
another one. What we said like we sent like
51:54
we said like we just combine all of them into one.
51:58
So now we have 1 pipeline which process the data and it drops
52:03
unnecessary data. It also builds the kind of state
52:07
of the user like whether that user has book a ride, whether he
52:13
got it right or not, whether he dropped off.
52:15
So we have maintained the state of the user similarly for both
52:20
like passenger and driver and based on that we know the
52:26
current state of the world and we have all these metadata
52:32
stored and immediate every minute or so.
52:36
So the downstream processing that is basically aggregating
52:39
all these data, they can get all the relevant information and
52:43
then they can aggregate it. And so the aggregation logic is
52:48
different for different features.
52:49
So you can have a different pipelines.
52:52
Does it make sense or? So what I was saying is like,
52:55
whatever the common processing, we extracted out and we
52:58
centralized it. Yeah, but we separated out other
53:02
pipe pipelines where their processing logic was different.
53:07
And that's how we were able to cut, cut cost and centralized
53:09
the processing. The other thing that we did is
53:12
like writing code was kind of used to take longer time.
53:17
What we did is like we came up with the YAML based pipeline
53:21
generation where you do not need to write code.
53:24
You write a kind of YAML configuration file and it does
53:28
the processing for you. So that cut down the development
53:32
time as well. Tell me more about.
53:35
That So what we did is like we realized that you know if we
53:41
have to build a new feature most of the time we are doing very
53:46
similar processing so let's say if you want to figure out
53:51
whether this is a driver or passenger, that is 1 function
53:57
right left. So what we did is like we find
53:59
all these common functionality, we provide that SDK and then.
54:04
Yeah, OK, I get it. You would put it into the YAML
54:09
as something that people could just use.
54:11
They can define it's already defined and then wow, OK.
54:17
So we said like, OK, you know, I do not want our data scientist
54:22
to understand the whole thing, like how to write a pipeline.
54:26
I'll say that, OK, I'll define a very simple way of defining A
54:29
pipeline. So I will create all these, you
54:33
know, general purpose functionality for you and then
54:36
you can just declare this in a YAML file how you want to
54:41
process. So they do not need to
54:43
understand how to define the windowing, how we're doing work
54:48
and all. They just need to tell like I
54:50
need this one minute long window or two-minute long or whatever.
54:53
And then just define that in a YAML and our system will read
54:59
that YAML file. Look at like how you are defined
55:02
all pipeline and will build the pipeline for you.
55:05
Yeah, that cut down our development time.
55:09
No. You did mention that certain
55:11
pipelines have 5 different features being created with
55:16
them. You're getting like 2 for the
55:18
price of one on certain pipelines, right?
55:21
Is it that they have certain steps that they're doing
55:27
together? Like, there's these certain
55:28
functions and so you can bundle them up, but eventually they
55:34
diverge, yes. That's right.
55:37
Like you, So the common functionalities they are a kind
55:41
of margin and centralized, the specialized zone still brashed
55:45
out and that's how like we are able to save it earlier like all
55:49
these pipeline was kind of independent but you know we were
55:52
wasting other sources with yeah, OK, let's combine the
55:55
centralized pieces. Dude, this has been awesome.
55:58
Is there anything else that you want to hit on that we didn't
56:01
talk about? I think we have current most of
56:04
it. But yeah, I would like if you
56:06
have any other question, just let me know.
56:08
I'm just so glad you didn't say, oh, we found a way to cut down
56:12
on our coding time. I thought for sure you were
56:14
going to say, we now use cursor and AI code generation.
56:18
And I was putting, I was like, please don't say that.
56:21
It's some of them are like precursor as well.
56:25
So when we initially introduced the pipeline, right, we were
56:29
able to cut down almost like 60% of the code base.
56:33
Wow, because of the streaming like streaming provide all
56:36
these, you know, processing what we call operators or what we
56:41
call functionality where you do not need to write those code
56:44
like let's say if you want to join anything, you can just
56:47
write one operator. It does that work for you.
56:49
Like it hides all the what we call implementation details from
56:53
you. So if it's pretty simple then
56:56
what we did is like we further went ahead and said, OK, even I
56:59
do not want to define all these operator like what is the
57:02
easiest way to do it? And we came up with a
57:04
declarative YAML based pipeline, so where you just need to write
57:07
one line or like for one operator you just need to
57:11
declare A1 line configuration. That's all.
57:14
That's so smart. That pipeline is not more than
57:18
or like that definition is not more than 2030 lines of code
57:23
rest of the other rest of the other like processing that is
57:27
like hidden by our platform. It is completely abstracted out
57:31
so they do not need to worry about it.
57:33
So now you can think about where we had thousands and thousands
57:36
of line of code for processing one feature, then we cut it down
57:40
to almost 30 or 40 lines of YAML based configuration.
57:46
That's why you get paid big bucks.
57:48
I'll tell you what, that is a genius implementation.
57:51
I like it.