>> Hi, everyone. With this release of Citus,
we really put the focus on the ease of scaling out.
Regardless, if you start from a single basic tier cluster,
which means one coordinator node and no workers,
we want to be absolutely sure that you will be able
to quickly scale out without any pain points,
even if you made some bad decisions or forgot to do some things at
the beginning of your journey with Cosmos DB for PostgreSQL.
Let's start with a simple table Observations
which apparently was created by me
beforehand so let's drop it and create it again.
Just so we make sure I don't have
any leftovers from me trying this before.
We have this simple table
"Observations" that we will fill with some data.
It's just several fields, Tenant Id,
Measurement, Payload, Observation time.
We want to put some data in it.
But where to get data?
One of the things that we want to ease is not only
scaling out but getting your data into the Cloud.
A common pattern of getting data
into the Cloud is to use a service like
Azure Blob Storage to use it as
a staging environment or as a target environment for your,
for example, IoT devices derived data to and
then directly pulled in into your Citus cluster.
We enabled an extension in Citus,
which is called Azure Storage.
We can see it's already installed in
this case. We can see it here.
We can use this extension
to operate on existing blob storage containers.
I prepared a container beforehand,
which is named PGABS,
which is PG for Postgres ABS for Azure Blob Storage.
It's a public container.
Public is just a container name.
Its accessibility level is set to
container public because this allows me
to access its contents without adding any access keys.
Though you can also do that if you have a private container.
We will use the blob list function from
the extension to show empty file in this public container,
how much storage it uses,
and its MD5 Checksum.
Calling this, we see that this container has several files.
We will be pulling in observations-1.csv.gz,
which is a comma-separated values file with a zip compression.
It weights in around 11 megabites of data.
Let's pull that in using the Copy command
and you can do this on
normal PostgreSQL without the Azure Storage extension,
you can install the extension, do it.
But with our extension,
the database recognizes the blob storage URLs,
and is able to do
API/SDK calls to obtain
the data and directly pull it in into your database.
The moment we execute this command,
a connection is made to Azure Blob Storage.
The data is fetched over the network between the services,
then compressed and ingested into the table
itself so that you don't have to copy it anywhere beforehand.
You choose directly ingest it from the Blob storage.
You can see that we loaded
three million records into the Observations table.
We can quickly show that we have all the data inside.
We can just show some records of
it and we have some pseudorandom data.
Now we can also check the Citus tables.
Citus tables shows you
the actual distributed tables that
the system knows about and we don't have any.
One of the frequent beginner mistakes,
or perhaps not even a mistake if you start in
the basic tier and didn't think that you will
need to distribute this table because it started
out small and you didn't anticipate it.
But then suddenly you realize that "Hey,
I need to distribute this table because I want to scale it out,
which needs much more data than I anticipated."
In the base, if you forgot to do that
and had a lot of data in
the table and tried to distribute your table,
then you would end up with a blocking operation.
This call will be blocked.
Let's show it.
I will open a second connection to
the same cluster where I will try to
insert some data into the table while I am writing to it.
As you can see,
the insert on the right hand is blocked and
waiting for the operation on the left connection to finish,
and only after it's finished, I'm done.
This means that your application
essentially is not performing rights,
which might be very problematic from the business perspective.
In Citus 11.11,
we introduced a new functionality
called Create Distributed Table Concurrently.
I will drop the Observations table
again and it's closed its connection to see someone.
Let's recreate our table the same way as we did before.
I will load the data slightly
differently this time around just to show you
another API function from the Blob Storage extension.
Here we are doing an insert into
select from Blob Storage, Blob Get,
which allows me to fetch a single file,
and as you can imagine,
this allows us to do
some more on-the-fly manipulations of the data.
For example, if I wanted to,
I could change or cast some field on the file.
We are just showing it to show you the flexibility implemented.
Now, we will do the same thing.
We will distribute the table but we will do it concurrently.
Now, let's get some space here and connect
to the cluster again and prepare our insert operation.
Now I will first show you
that this table is simply has some data in it.
That's the three million rows that we saw beforehand
and we used the sampling so create distributed table concurrently,
the name of the table,
and the column that we want to distribute on.
In Citus, you have to tell
the database what's your distribution column,
the value of this column will be then used to decide where
the data based on the value
goes through to which chart and on which node.
I'm going to execute this now and while this is being done,
I'm inserting rows in the second tenant.
You can see the operational on the left is still
ongoing while your application can perform,
insert statements into the table, and other modifications.
This essentially means that this
is a safer operation that can be done
online without a lot of disruption in your business.
Let's close this.
Now the table is distributed.
We can see that there are more rows inserted,
so the three million that were there and the 20 rows that
were inserted during the distribution.
>> We can also now take a look at the data.
I mentioned, the citus tables.
We can see that now observations is present there.
It's a distributed table on tenant id,
it waits around 311 megabytes
because the data we ingest it was gripped.
When we pulled it over with
the Azure Storage extension to
extension to compress the data for us on the fly.
Let's see how the data looks.
We can imagine that even when you did that,
some of your penance might be not very comparable to others.
You might have a big company loading a lot of data
and a number of small customers that store a fraction of that.
Let's look how our distribution looks like here.
By looking at the data itself and grouping it by tenant id,
we can see that one tenant with id 293
has significantly more records
than all the other tenants present in the database.
Now, we will find the shard id,
the identifier of the shard that
this specific tenant is assigned to.
This is shard 102093.
Now we can look at the shard sizes.
Let's see how even the distribution is.
You can see we have a lot of shards.
One shard is standing out of 78 megabytes,
and this is the shard 102093,
as we talked before,
I can scroll up to show you.
We can see that's the same shard id.
We know that this is taking significantly more data.
It's skewing the data into one shot.
How are they balanced per workers.
The cluster I'm demonstrating on has
one coordinator node and two workers.
I can do this query to show that for each node,
for each table, what's
the size of the shards assigned to this node?
We can see that the Worker 1
is significantly larger than Worker 0.
That's one situation where we for example would like to take
this large tenant assign
him or assign this customer to its own shard,
and move that to a new worker or the balance it either way,
isolated from other customers because
this specific customer has larger needs.
We can do this in citus and in citus 11.1,
this operation is non-blocking,
like creative people concurrent.
When we are doing this operation,
we will again perform the inserts to
the observations table while
distributing the tenant to a new shard.
The distribution is now
ongoing and we are inserting new observations.
Again, your application can continue to work.
Serve their business,
while we are fixing the problem in our initial architecture,
or we are reacting to changing requirements because businesses
change in disc like might have
changed its user to patterns recently,
so you were not able to anticipate that.
Now the tenant is in a new shard and we have
the shard id provided to us.
Let's reclaim some space so we can see more.
We can look again at how the shard sizes are.
This is the new shard, the 102110.
Then let's confirm that it matches with the one here.
Yes. We can see that this specific shard is now 74 megabytes.
Now, nothing in the distribution self changed
between the worker nodes because we
just moved the client to a new shard,
we also would like to do something with this shard.
Now let's look at the balancing again.
We see there is no change here.
Let's surround rebalancer.
Before citus was 11.1,
we had to maintain a p-SQL connection to the cluster,
round the rebalancer,
and wait for it to finish.
Now with citus 11.1 we introduce
a new API and essentially a new framework of running jobs in
the background that you can simply kickoff with and
connect back to your cluster and it still would finish,
you don't have to keep this connection open.
As you can see, we have this new pg dist background task and
pg dist background job tables,
which we can use to look at the running jobs.
Let's look at the second job,
as we have suggested here by down click.
We can see that we had task id 2 started by citus, it's done.
It shard placement had moved one of the shards to another node.
We can look at the shard placements between the nodes.
We knew it was 69 megabytes on Worker 0,151 megabytes on Worker 1.
Now it's a bit more even.
It's 73 megabytes on Worker 0,146 megabytes on Worker 1.
Now, it won't be perfectly even because there's no way to,
the sizes of the shards doesn't
allow a perfect split in this case.
But if you added more nodes,
the data could be more evenly spread out among the other nodes,
or you can use some other citus features
to move the large tenant to its own node.
That's not done the thing we did with the rebalancer
by having it run in the background, it's more reliable.
It can restart on database recovery. It's faster.
We spent a lot of time to improve
its performance and they start to serve you as best as it can.
Without anything more to add.
Thanks for watching my demo and have a great event.
[MUSIC]