ALEXA GRIFFITH: Hi, my name is Alexa Griffith.
Yes, my name really is Alexa, but hopefully I
won't echo today.
And I'm presenting with my amazing [INAUDIBLE],,
Mike Hurwitz, who we call Danger.
We're both Software Engineers on the Data Science Infrastructure
Team at Bluecore, and today, we're
going to talk about how we use Google Cloud to serve
tens of thousands of personalized
recommendations per second.
At Bluecore we have a saying, "As simple as possible,
as powerful as necessary."
This is one of our company values,
and we will keep this in mind as we discuss
the evolution of building our robust recommendation
service throughout this talk.
So GCP has enabled us to build a service that
has this value of being as simple as possible,
and as powerful as necessary.
In building the recommendation service,
we're able to leverage the wide variety of GCP tools
to store, query, and serve data.
They all [INAUDIBLE] some purpose,
but really, it's the breadth of their offerings
that makes services like ours possible, while allowing
us to maintain a low operational overhead.
So you might be wondering, what is Bluecore?
Bluecore offers intelligent e-commerce marketing
as a service.
We send you marketing emails that don't suck,
and we aim to send you better emails, not more.
We're a company that's around 7 years old.
We have around 250 employees.
We serve hundreds of large retailers,
including some you may have heard of,
like Nike and Sephora.
What makes us different is that we don't send the same email
to every person.
Almost all of our millions of emails that we send every day
are personalized.
Additionally, we've started providing
real-time recommendations on retailers' websites.
And we're going to refer to this as our on-site product.
So let's discuss.
What even is a recommendation?
Well, when we talk about a recommendation,
we're talking about products that we
think someone will like based on either a product they've
interacted with, or they may have a preference for,
given their purchase and viewing history in the past.
Essentially, we map either an email or a product
ID to a list of recommended products.
For example, in these cases, two recipients
of the same campaign, which is just the type of email send,
left different items in their carts.
And we call these "abandoned cart emails."
If we take a look at the email on the left,
the person's left jeans in their cart.
It's possible that, based on their past purchasing
and viewing history, we want to recommend
to them a cami or a t-shirt.
On the other hand, in the right email,
a person has left a collared shirt in their cart.
We may recommend other products similar to the collared shirt
as products the user may also like.
For example, here we recommend other types of collared shirts.
But there are some things to consider.
It's not just as easy as putting some pictures in an email
and sending the email.
There are certain things we need to take into account,
like product exclusion rules, which are evaluated at runtime.
What a product's exclusion rule could look like is you say,
in the email, you don't want to show a product that's
on discount or out of stock.
Or maybe even you want to show a product that's
recently come back into stock.
Since products update constantly,
it's very important that we have an accurate catalog
and don't show invalid products in the email.
This requirement definitely makes our service
a bit more interesting.
And we'll talk about how we use our Product Update
Listener to keep our product catalog fresh.
MARK HURWITZ: So a marketer's decided to send an email,
and they create a campaign.
The campaign has an audience and a template.
A template has blocks, which may contain recommended products.
The template is mixed with the audience query that
runs in BigQuery, and with those, we
can generate the emails.
Products are then joined with the audience in the query,
which means that any decisions the template makes cannot rely
on anything that the query is unaware of.
Generating audiences this way is very flexible,
and we consider it to be a competitive advantage.
But applying recommendations at this stage is very limiting.
So you may be asking, is this good enough?
On-site was a new frontier for Bluecore
that forced us to re-evaluate our architecture.
Latency became a primary driver, not just bandwidth.
Audience generation and personalization
are separate, which means the templates are
limited to data appended to the rows of the audience query.
We want to support templates that are structurally dynamic,
rather than just filling in blocks.
And that means either making the audience query aware
of all of the decisions that need to be made,
which is pretty difficult, or allowing the template
to pull the recommendations it knows
it needs at runtime, on a per-email basis.
A/B testing is another example of a personalized runtime
decision that's difficult to make in the database.
Previously, we were copying some recommendations into Datastore
to support these use cases.
And while Datastore is great, this is really not
what it was made for.
And it got to be very expensive.
So we had to build something new.
Pages on the web start to feel slow
at 300 to 400 milliseconds, and recommendations is only
part of the on-site campaign process.
There's also generating the template,
and translating all that data to and from the browser.
So we gave ourselves about 100 milliseconds
to do all the work that we needed to do
on the recommendation side.
We want to be able to serve about 50
million emails an hour.
And if you do a little math, that
works out to 13,000 emails a second.
We targeted 20,000 emails a second,
because we figured that would be enough headroom.
Doing product filtering in the email
means that there must be enough products returned
that the filters will be run across all of them,
and there'll still be enough to fill in the blocks.
That means sending more products than are actually required down
into the templating engine.
We wanted to move those decisions up
into the recommendation service, so that we were only
returning the products that we needed,
rather than returning too many.
Or potentially, even worse, too few.
Supporting recommendations via A/B tests
is vital, because frankly, without testing
you can't improve.
And we want to do all of this without spending
all of our money.
ALEXA GRIFFITH: So for the naive implementation,
we want to change as little as possible.
We know this probably won't be good enough in the end,
but what it will do is it will give us a good idea of what
areas we need to improve on.
So if we want to change as little as possible,
how would we do it?
Well, we would choose to write a GRPC service to communicate
with our templating engine, and we
want to go ahead and make use of the data
that we already have available.
Our recommendations data already listed in BigQuery,
along with our products data that comes from Datastore.
So this would be the minimum we need for our service
to do runtime personalization.
The recommendation service makes up
one of three parts of personalization,
with the other two being templating, which
comes from Datastore, and the audience, which is just
all the people that receive an email
and is stored in BigQuery.
The recommendation service runs in Kubernetes' engine,
and is written in Go.
and the products data lives in Datastore,
while the recommendations live mainly in BigQuery, with some
copied into Datastore.
So now we've checked the "as simple as possible" box.
But do we meet the "as powerful as necessary" requirements?
Well, when considering if our service meets
these requirements, we're going to evaluate the recommendations
and products components.
Remember, we now have to consider on-site,
or the recommendations on the company's website,
as part of the requirements.
BigQuery is great for bandwidth, but now latency is our problem.
We're failing to meet our on-site latency requirements.
Even if we did export all of our recommendations
from BigQuery to Datastore, we still
might not hit our latency goals.
And keep in mind, for us, using Datastore is expensive here.
But we can't blame a hammer for being a bad screwdriver.
We just need the right tools.
So first, we're going to discuss how we can
liberate our recommendations.
As I just mentioned, BigQuery cannot be the source
of recommendations at runtime.
The latency is just too high.
So we looked into Bigtable instead.
Why Bigtable, you may be asking.
Well, caching is not a good option for us
because the specific recommendation data for a given
recommendation type, email, and namespace,
are often only read once before they're
invalidated in the cache.
And a cache miss would mean reading directly
from BigQuery, which costs us in latency,
or Datastore, which is expensive.
On the other hand, Bigtable latencies
are around 15 to 20 milliseconds,
and better the harder you hit it.
So if we consolidate our data into Bigtable,
we can meet our latency's requirements.
Our recommendations, including events,
and our generated recommendations from our data
science models, are still stored in BigQuery.
Additionally, we started generating
some of our recommendations using data proc,
and these are stored in GCS.
We created a service written in Go
that's a Kubernetes task running in Airflow,
just to migrate and consolidate our recommendations
data into Bigtable.
The GCS and BigQuery storage API made loading data
into Bigtable fairly easy.
With a small Bigtable cluster of just three nodes,
we're able to load, read, and write
150,000 to 200,000 rows per second.
So now that we have our data into Bigtable,
let's talk a little bit about how we deal with our data
in Bigtable.
We have hundreds of namespaces, but Bigtable has a 1,000-table
quota.
We can't have a per-namespace table for each
of our recommendation models, or we're going to quickly reach
that 1,000-table quota limit.
That's OK, though.
They don't call it "Mediumtable" for a reason.
This is what Bigtable was made for.
So instead, we have a per-table--
we have a table-per-recommendation model,
and our schema is the key as the partner and ID,
which ID could be email.
And the value is a list of product ID's.
So just to recap, this is what we had before,
paying attention to the recommendations coming
from BigQuery and Datastore, which are now
consolidated into Bigtable via our BigQuery-to-Bigtable
service.
And we picked up literally seconds of latency.
So now that we've improved the recommendation
component of the service, are we good now?
Are we meeting our on-site latency requirements?
Well, I mentioned previously, on the previous slide,
that there are two areas where we can improve
our recommendation service.
The recommendations portion and the products.
Now that we've improved recommendations,
we're still constantly fetching products and Datastore,
causing it to burn a hole in our wallets.
Additionally, we're failing the on-site latency requirements
because of this too.
So we measured Datastore latency from a GKE application written
in Python, and each of the products
we fetched from Datastore are around 2 kilobytes per entity.
We found that the Datastore latencies aren't great.
They're about 45 milliseconds to get started,
and then an additional 4.2 milliseconds per key.
So like I said, not great.
But they aren't totally disqualifying
for our use of Datastore.
MARK HURWITZ: Using Datastore didn't leave us
a lot of headroom.
So if Datastore isn't going to get us where we need to go,
we need to find something else.
What we need is a latency smack-down!
Our objects are about 2 kilobytes,
and so these are cachable.
We found Memorystore to be about two orders of magnitude
faster than Datastore.
Fetching a single key out of Datastore
cost about 50 milliseconds.
But Memorystore, for the same load,
costs only 125 microseconds.
We're usually reading about 20 to 40 products per request,
so these numbers are more realistic than you might guess.
Because Memorystore is managed Redis,
it was easy to find clients that support
all the features that we need for performance,
such as pipelining.
We need to avoid the calls to Datastore,
which means we need to have a long TTL on our cache.
But Alexa already told us the products are constantly
changing.
They're going in and out of stock.
Prices are changing, images are changing.
So we can't have a long TTL.
Our product can just pipeline right to Datastore,
as it always has.
But now, it's also notifying us of invalidation
by pushing product ID's onto Google Cloud Pub/Sub.
We have a listener that reads those ideas from Pub/Sub,
pulls the products from Datastore,
and writes them to our cache.
You may be wondering, why do we send ID's instead of just
sending the product bodies?
Well, the product [INAUDIBLE] is concurrent,
so we can't know of any single update as authoritative.
By sending ID's instead of bodies on Pub/Sub,
we can guarantee that the cache is eventually consistent,
rather than being potentially inconsistent forever.
And we can have a very long TTL.
Datastore natively returns protobufs,
and those come out to about 2 kilobytes, on average.
Now, we could have come up with a new serialization format that
does everything that Datastore's native serialization
format does, but there was really no reason to do that,
because it's already doing it.
But with about 2 kilobytes per product,
and many, many products, we'd end up
with a huge cache, which is both expensive and slow.
So, how do we mitigate that?
The first thing is that we don't update products that
aren't already in the cache.
Products get into the cache by being
used in a marketing product, which
means that the only objects that'll be in the cache
are the objects that are actually being sent out.
So if a partner of ours happens to have
a catalog of a million products, but they're only sending out
emails with 1,000 of those products,
we only have to have 1,000 in our cache.
The second thing we looked into was compression,
but compressing small items is hard.
Most compression algorithms learn as they go,
which means if you're compressing a book,
the first page is going to compress significantly worse
than the 10th page or the 100th page.
And with only 2 kilobytes, there really isn't a lot of time
to learn.
By the time you get to the end of the product
and finally know how to compress it,
there's no more to compress.
So what did we do?
Well, first thing is that we downloaded 1,858 products
from Datastore for one of our partners,
which came out to about 32 megabytes.
And we then tried to compress them using gzip and xz.
And while the LZMA compression of xz
is really too slow for runtime, that's
kind of OK, because we're really just testing here.
So we then found that we were unable to do better
than 2-to-1 compression, even with the very slow xz.
Well, that's not really good enough.
So how good could it be?
Well, we took those the same 1,858 products
and we tarred them all together, effectively making one
big book, and compressed again.
And we found that gzip was able to do better than 5-to-1,
and xz was even able to do better than 8-to-1.
But we can't just tar these products at runtime.
That doesn't make any sense, because we
don't need all the products.
We don't know which ones we need.
So we needed something else.
Fortunately for us, Zstandard exists,
which allows you to train its compression ahead of time.
They call this "dictionary compression,"
and that's exactly what we do.
On a per-partner basis, we download
a sample of the products, compress them,
training the dictionary.
And then we store those dictionaries into Datastore,
caching them in the service as well.
So any time we read or write a product,
we know exactly what dictionary to use.
And the compression is fantastic.
As you can see here, Zstandard with dictionary
is better than 5-to-1, even on small, 2-kilobyte [INAUDIBLE]..
There are other high-performance and decompression algorithms,
such as Snappy and LZ4.
However, those don't offer dictionary compression, which
as we've seen, is a big factor with such small bodies.
As a reminder, this is where we were before,
and this is where we are now.
We have products still stored in Datastore, as we always have.
But now they're being kept up-to-date
in a cache with our Product Update Listener,
which is receiving invalidation messages over Google Cloud
Pub/Sub.
The recommendation service itself has a small LRU
cache with a short TTL, just to keep the traffic-to-memory
stored down to a dull roar.
And we think we're finally as powerful as necessary.
But, how to be sure?
The only way is to test it.
And fortunately for us, the power of GKE,
this was pretty straightforward.
We were able to deploy our service,
and to the attacking client, into Google Kubernetes Engine.
And we could then scale both the service and that client
with a single command.
The entire run, including re-scaling the components,
was completely scripted and hands-off.
And how did we do?
Well, we did pretty well.
With 16 two-core pods running the service,
we were able to pull between 56 and 60 kilohertz, while still
staying within our latency budget.
And we hit our target of 20 kilohertz with just four pods.
Yes, we are finally as powerful as necessary.
Now that this code is out in production,
well, changes need to be made, and we need to live with it.
One thing that we do a lot is--
when changes are made, we'll deploy a canary
into Kubernetes.
That allows us to validate that the new code doesn't
have a regression, in terms of correctness or performance.
And then once that's proven, we can scale it up
to the rest of our pods.
Logs are exported automatically from GKE into the BigQuery,
and so we've got a one-stop shop for when we need
to look and see what happened.
And we're exporting all of our metrics to Datadog.
And because this is a production service,
we've got PagerDuty to wake us up in the middle of the night
if something catastrophic should happen.
So, what have we got?
Well, recommendations are being written
to BigQuery and Google Cloud Storage,
as they always have been.
But now they're being copied into Google Cloud Bigtable.
Products are stored in Datastore, again,
as they always were, but are now being cached in Memorystore
and kept up-to-date using invalidation triggered
through Google Cloud Pub/Sub.
Our clients hit us using either protobuf over HTTP/1.1,
or gRPC over HTTP2.
And the whole thing is deployed into Google Kubernetes Engine.
I'd like to thank you all for listening,
and thank you to Google for giving us the opportunity
to speak to you today.