CLAIRE GIORDANO: This demo is created by Marco Slot who is a principal engineer on the Postgres
team at Microsoft. It's about HTAP, hybrid transactional
analytical processing. Using Postgres and Hyperscale (Citus) which is now available
on Azure, built into our managed Postgres service.
Now Citus is an extension to Postgres and not a fork and it's open source and
what Citus does is transform Postgres into a distributed database, distributing
both your data and your queries across the cluster, which gives you parallelism, and
lets you use all the CPU, memory, and disk of the distributed cluster. And with that
I want to introduce Marco, who will tell us a bit more about the demo. MARCO: I will show
a demo of a database for an order processing system for warehouses and
it's all going to be running on Azure and the application is simulated by an
awesome tool called HammerDB, which is an open source database benchmarking
tool which supports several different databases and several different
workloads and one of the workloads it supports it's called TPCC
and it's a standard benchmark for measuring transaction performance and
that models an order processing system for warehouses. And so we're going to do a
side-by-side comparison of a Hyperscale (Citus) cluster and a single Postgres server
running transactions using HammerDB and running analytical queries and then
finally I will show you how you can use roll-up tables to do millisecond
analytics in the same database. CLAIRE: OK, I think it's time to jump in. MARCO: So let's switch to my screen, cool. So I'm going to
do an HTAP demo and HTAP might sound like something very new but almost all
applications have transactional and analytical components, I mean who runs an
online shop which processes millions of orders per day but never asks how much
money they're making? And a relational database like Postgres is quite suitable
for handling both transactions and analytics, but when your database becomes
very large, a single-node database might
struggle to answer the analytical queries and you might end up deploying a
separate system just for analytics. But with a Citus cluster you can keep up
with very large transaction volumes and also do very fast analytics on the same
database. Now the easiest way to create a Citus cluster is to go to the Azure
portal to provision an Azure Database for PostgreSQL database, and when you
create a new database you can choose the Hyperscale (Citus)deployment option and
when you click create that will take you to a form where you can pick things like
the region and the number of worker nodes, but in this case I already created
a cluster. So here you see some details about my cluster and on the right you
can see this is a cluster with 10 worker nodes with 16 cores each, so that's 160
cores in total. And if I ever need more capacity than that, I can go to the
configure tab on the left and use the slider to pick my new number of worker
nodes. I'm going to use the command line for this demo because, well, I love to
use command line and it also gives me the opportunity to do a side-by-side
comparison of my 10 node Hyperscale (Citus) cluster which will be on the left
and a single Postgres server with 16 cores that will be on the right.
So the Hyperscale (Citus) cluster is 10 times bigger than the Postgres server. So
let's have a look at what's in these databases, and I can see that HammerDB
generated a table of customer data and orders and the order_line table which
contains a record for every item ever ordered, and we'll do some
analytical queries on this table later. Now on Citus on the left, you can see
that this adds up to total of over 300 gigabytes and on the right on Postgres
we have around 200 gigabytes, the reason there's more data in Citus right now is
because both these databases are processing transactions and Citus does
so at a higher rate, so the database is growing faster. Now let's have a look at
what's going on in these databases by querying pg_stat_activity, which shows me
all the connections to Postgres and what query they are running. And you
see that both databases are handling hundreds of concurrent stored procedure
calls. Now stored procedure is basically a little program containing queries and
data modification statements that's stored in the database and typically
runs as one transaction. And we can even see the source code of the program for
example let's look at the payment procedure and open it up
and you can see that it has SELECTs and some UPDATEs and an IF
statement with some SELECTs and then some more UPDATEs and finally an INSERT
and so this all runs as one transaction. And we're doing the same thing on
Postgres and on Hyperscale (Citus), but because this transaction only deals with
a single warehouse, on Citus we can route it to the different worker nodes
and speed up the transaction throughput. So in the background I'm tracking the
number of transactions that each of these systems is running per minute and
I can see that using the transaction throughput view that I created. And if I
look to the right, to the Postgres server, I can see that in the last 10 minutes
it did between 40 and 50 thousand transactions per minute which is quite
decent because I could run a pretty large online shop with that. But on the
left, the Hyperscale (Citus) cluster was able to run around 900 thousand
transactions per minute so that's around 20 times faster. And the reason it's
twenty times faster and not just 10 times faster even though it has ten
times more CPU and ten times more IOPs, is that it also has ten times more
memory and therefore it can fit all the data in memory. And I if I want my data
to continue to fit in memory, I could even keep adding more worker nodes over
time, so this is also a big difference between a sharded architecture and an
architecture that uses read replicas, because in read replicas you typically
have the same subset of the data in memory on every node whereas in a
sharded architecture we divide it across the nodes and all of the nodes can also
handle writes, so that makes it quite powerful. So far, we've looked at the
"T" in HTAP, now let's look at the "A", the analytics. So here I'm running a
query that calculates the total amount of sales by state and gives me the
top 10. And the data here is completely made up, but in the bottom left you can
see that Citus finished the query in around 10 seconds. On the right, Postgres
is still working on it and unfortunately we will not see it
finished during this demo or even during this talk. I know this because I actually
already ran the query on the Postgres server before I started the demo so I'll
switch to a different screen and here on the right you can see that it took 53
minutes to complete the query on the Postgres server, so the Hyperscale
(Citus) cluster was over 300 times faster at answering this analytical query by
parallelizing across all the cores in my cluster. And if I do a SELECT count(*)
from order_line, I can see that this query was actually a join on over 700
million rows as I can see in the bottom left corner. So this shows that Hyperscale (Citus)
can do these analytical queries on top of of a transactional workload, in just a few seconds.
But what if I need it to be even faster? So we'll leave the Postgres server behind, and
focus on the Hyperscale (Citus) cluster. What we often recommend to Hyperscale
Citus customers who need very fast analytics is to create roll-up tables
and a rollup contains pre-aggregated data for some
set of dimensions, for example here I have a rollup called total_sales_rollup
and for every warehouse, sales district, and state, it stores the total_sales
amount so far. Now we'll need to update the rollup as new transactions are
happening and for that I created another stored procedure. My stored procedure
contains 1 command, an INSERT..SELECT command, which is a command that INSERTs
the result of a SELECT query into a table, in this case it INSERTs into my
rollup table, and it finds all the unprocessed orders that have not been
added to the rollup yet, and for every combination of warehouse, sales district,
and state, it calculates the sum of the sales and inserts that, or if a record
already exists it does an UPDATE by adding to the total_sales amount. On a
single Postgres server, this command would not be able to keep up with the large
transaction volumes because it can only use a single core, but the Hyperscale
Citus cluster is always able to keep up by parallelizing this command across all
the cores in the cluster so Hyperscale (Citus) can not just parallelize SELECT
commands but actually any type of Postgres command. And because we're only
querying for the unprocessed orders and the worker nodes can use indexes to find
those orders efficiently, we can combine the power of parallelism with indexes in
one big distributed transaction that takes all the new data and updates my
rollup table. Now let's run the procedure to catch up with all the
latest transactions,you can see this only takes around 2 seconds. Now we can
query the rollup table, and because this table is relatively small and has
indexes, it only takes around 20 milliseconds to get the same query
results that I got with my analytical query before, except it's a bit more
up-to-date now. So I went from a query taking 53 minutes on a Postgres server, to
using parallelism in Hyperscale (Citus) to make it 10 seconds, and then by doing
a parallel distributed transaction inside my cluster I was able to bring it
down to 20 milliseconds. So what we've seen is that Hyperscale (Citus) is able
to handle and scale transactional workloads really well because it's built
on Postgres and can just route the transactions to different Postgres servers.
And by parallelizing queries on top of many Postgres servers it's also able to
handle analytical queries at the same time on very large data sets and then by
combining transactions with parallel queries we can actually process the data
inside the database at a high rate to make analytics even faster and this is
really the power of HTAP. Normally I might have to deploy several systems to
get all these different functionalities to be that fast but with
Hyperscale (Citus) I have a database that can scale and will handle almost anything.
CLAIRE: Will all applications see the kind of impressive performance increases you just showed us?
MARCO: Well, there is of course question of what are we
comparing, and it shouldn't come as a surprise that a 10 node Citus cluster is
faster than a single Postgres server, but it might be a little surprising that we
could make it 150 thousand times faster in the end. And actually you
might still be able to get that if you used an architecture where you took the
data out and put it in an analytics database and had some cache to
make make the roll-up query fast, but I didn't have to do that. All I had to do
was run a few SQL queries and they run in parallel and do all the work
for me. And I mean the technique of creating rollup tables is really very
powerful for scaling real time analytics and for the transaction throughput we
used a technique called stored procedure delegation. And to make these techniques
work well, one thing that's really important is to find a good distribution
column, ideally it's one if you have multiple big tables that is common to
all tables, such that if you have a transaction it can touch on
different tables but on the same distribution column value. Or if you do a
JOIN between two tables ideally that JOIN is on the distribution column
value. And very often it's actually not that hard to find this column for
example it could be or your customer_ID or your device_ID, and if you can find
that column, then you can scale your application in the same way.
CLAIRE: At the end of the demo Marco, you referred to Citus as an almost-anything database. What did you mean by that?
MARCO: Well remember that Citus is an extension to Postgres, it doesn't take any
functionality away. And as a relational extensible database Postgres is capable
of so many things already on a single server, there are so many different
applications that that run on Postgres, but the one thing it cannot do well is
scale to meet the needs of very data-intensive applications. But now with
the distributed tables that you create using Citus, it can do that too. So
Hyperscale (Citus) transforms Postgres into a database that can really do almost
anything. CLAIRE: So let's say that we have some
developers watching this talk and they want to get started,
what do you recommend, what are the best ways to roll up your sleeves and try this
out. MARCO: Well it's easy to download the open source packages from our websites and
set it up on your on your local Postgres server on your laptop, and once you've
done that, I recommend going through some of the tutorials on our documentation, we
have for example we have a tutorial on how to distribute a
software-as-a-service a multi-tenant application using Citus, and another one
on how to build a real-time analytics dashboard using Citus, so I would
recommend going through those tutorials and learning about the distribution
column and then take it from there.
CLAIRE: Do you have a favorite blog post?
MARCO: Yeah I think my favorite is one you wrote about one of the biggest Citus
clusters out there. At Microsoft the Windows team has a cluster that
collects and analyzes all the telemetry data coming in from from Windows
machines and there's over one and a half petabytes of data in this cluster.
What's also really cool is that if you look at his architecture it really
speaks to this HTAP model where it pretty much is just Postgres with Citus
and an analytics dashboard and within the cluster, he does, the team does
transformations of the data, they do updates of all their data, and then all
these analytical queries on top of that. And the engineer who built it, Min Wei, he
tried all sorts of different solutions to building this system but
eventually he realized that Citus was the only way to do it, and he gave us
this great quote that "Distributed PostgreSQL is a game changer"
and I have to agree.