- [Pathik] Hello, my name is Pathik Shah.
I'm a senior big data architect
on the Amazon Athena team at AWS.
In this video, we'll be talking about
visual query execution analysis in Amazon Athena.
I'll briefly introduce Amazon Athena service
and then we'll discuss the new visual query analysis tools,
followed by a demo.
At the end,
I will share some resources to help you get started.
I'll start by sharing a few quick things about Athena.
Athena is a serverless analytics query service
that makes it easy for you to analyze your data lake.
You can query your data
without having to set up and manage
any servers or data warehouses.
Just point your data into Amazon S3, define the schema,
and start querying using the built-in query editor.
With Amazon Athena,
you pay only for the queries that you run.
You can save from 30% to 90% per query
and get better performance by compressing, partitioning,
and converting your data into column formats,
like Parquet or ORC.
Customers running queries in Athena
want to tune their queries
to reduce the overall execution time and the data scan.
They use EXPLAIN and EXPLAIN ANALYZE
to study the query execution plan and computational cost.
To make it easier to optimize the query,
we have launched visual query analysis tools.
So let's talk more about it.
Query Plan helps understand how the query will execute
before it's run.
(indistinct) Athena support
a generating EXPLAIN planning text,
JSON in a graphless format.
Customer decide to parse the text or JSON output
using custom tools to generate a visualization
for better understanding of the query.
Now, with a single click, you can generate the query plan
in either logical or distributed forms
on the Athena console.
This helps validate the SQL statement
and makes it easy to visually understand
the flow of the query.
Later on in the demo,
we will see how easy it is to see the joins
used in the query or validate predicate,
push down to the source,
or the operators that will be used in the query.
Once the query has executed,
the total runtime and data scanned by the query
is displayed on the console.
But customers are interested
in learning more about the query execution details.
To study computational cost for a Q1 query,
one of the options is to use EXPLAIN ANALYZE
which gives output in the form of text or JSON.
This output can be robust,
depending on the complexity of the query.
It can be difficult to read and parse the output.
To make this simpler and convenient,
Athena now supports displaying query runtime statistics
on the console after the query has completed the execution.
It will display this time spent
in the different phases of the query,
like queuing, planning, running, and service processing
along with the size of input and output data.
For queries that require deeper analysis,
you can now examine granular query execution details
presented as an interactive graph
that allows you to explore the flow of data
through our query.
You can use it to identify bottlenecks in complex queries,
inspect operators, and detailed execution statistics
for each query stage,
trace the flow of data between stages
to evaluate the impact of query predicates, and much more.
Let me show you a few examples
on how you can use the query analysis tools.
This is the Athena console.
I have created a couple of tables in advance
on the TPCS data set.
Here's a simple query that we'll run.
As you can see,
there's a new pattern over here called EXPLAIN.
Click on it, it will generate a visual plan of the EXPLAIN.
Let's go ahead and do that.
This is a logical plan of the query that I just showed.
Here, you'll see different operators
used in the query, and the flow of the operators.
(indistinct) to execute the query.
The first stage is a scan filter project
that is responsible for writing the table
and the details of the table can be found over here.
As you can see the table name is Line Item.
There are database name users, then users used as CSTP.
You can also see the predicate filters used.
So for example, in this query
we used (indistinct) and this is the filter value.
You can see the subsequent stages by rating
from downwards to upwards,
which is evident from the flow of TROs, as shown over here.
Now let's look at the distributed plan.
This shows an execution plan
in a distributed environment, which comprises of stages.
Each stage is processed by one or more nodes,
stage flow represents a data exchange
between cluster nodes.
Once you click on this stage,
on the right hand side you'll see the operators
that are used in this stage.
Let's look at the operators in detail.
So hit the expand button,
and you'll be able to see all the operators
used in the stage.
For this stage, we have only one operator,
which is ScanFilterProject,
but there will be stages
that would have more than one operators.
So this is a good view for such stages
where you want to look into the details
on the properties of the operators and stuff.
Stage zero is the final stage where output is gathered
and written back to S3.
Let's run the query.
As you can see over here,
there's a new tab called Query Stats,
which shows details on how much data was processed,
how many input rows were fed into the engine,
how many rows were part of the output.
For this query it seems that
there were around 600 million rows,
and the input size was 74 GB.
And the number of output rows produced were 248,000.
And the size of the output file is around 2.37 MB.
Also, you'll be able to see the query timeline,
which shows us different phase of the query
that is queuing, planning, execution and service processing.
Queuing is a phase
where query waits for resource allocation.
Planning is a phase where it drives metadata from the table
and schedules the task.
Execution is the phase where it performs
the actual operations on the engine,
and service processing is a phase
where it takes time to upload the results to S3.
For this query, like you can see,
the maximum amount of time was spent in the execution stage.
Now, let's say we want to see more details
on what was generated in the execution phase.
Let's click on Execution Details.
Now you can see the execution stage
comprises of a flow chart,
which shows different stages involved in the query.
Where a query is executed,
execution is divided into a hierarchy of stages.
Each query has a root stage,
which is responsible for aggregating the output
from other stages.
If you click on the stage, you will see different operators.
An operator consumes, transforms, and produces data.
For example, the ScanFilter operator
fetches data from a connector,
applies the filter predicate,
and parses the subject of the data
as input to the next operator.
The execution time shown over here
is the time taken to execute the stage.
This helps in determining which stage is expensive,
and in turn you can turn to the operator
using the stage to see
if you can optimize the operator usage.
For example, for this query,
we can see that maximum of time was spent in stage one.
So let's click on it.
On the right hand side you'll see the, again,
the similar stats,
like how much data was fed into this stage
and how much of data was output by this stage.
Along with the execution time.
You can also see the operator details over here,
which is similar to the one we saw in EXPLAIN plan.
Let's click the stage zero.
So this is a stage which is responsible for
aggregating all the outputs together
and then write it back to S3.
So let's click on expand all.
Now you can see here,
remote source means that the data was fed from stage one,
and then the next operator was the output operator,
which was responsible for putting two columns,
(indistinct) key and part key.
And as part of this, the input size was around 2.37 MB
and output size was also similar.
All right, so now let's see
how we can use these details to optimize our query.
For example, this is the query that joins two tables
and performs (indistinct) operation.
Let's hit the EXPLAIN plan.
Now you can see over here,
this is a distributed plan of the query,
but let's check the logical plan first.
As you can see over here, the query is performing join,
and you can also see what other tables
are being fed into the join.
For this query, in the ScanFilter operator,
we can see that the table that is being read over here
is part table.
And on the other side, there's a ScanProject stage,
which is responsible for reading table line item.
This layout tells that the table part
is on the left hand side of the join,
whereas table line items on the right hand side of the join.
It's for the best practices,
the corrector of join should be (indistinct)
such that the table with highest number of data
should be on the left hand side of the join.
The table with less number of data
should be on the right hand side of the join.
If you know your table sizes in advance,
this layout helps (indistinct)
that the correct join order is followed.
For the demo purpose, I used a simple query,
but if you have a complex query,
this layout will be useful
in validating the orders of the join is correct.
Then after the join stage,
there is aggregate partial stage,
followed by some (indistinct)
and then again, an aggregate stage.
So this is for counting the number of records
that is being output as part of the join operation.
And finally, there is an output stage,
which is responsible for writing results back to S3.
Now, let's study the execution details of this query.
As you can see over here, there are four stages involved.
Let's see the results of stage one.
The stage one comprises of inner join,
which gets data from a remote source two,
which is the stage two,
and remote source three, which is the stage three.
As you can see, the stage two is filling 20 million rows
and stage three is filling 600 million rows.
Now let's see what table is involved in stage two.
So in stage two, it's reading the table part.
And in stage three, it's reading the table line item.
Now, let's come back to the join order again.
As you can see, the table part is small over here,
since it's only outputting 267 megabytes of data,
whereas stage three has output 7.8 gigabytes of data.
So here, the order of join
does not follow the best practice,
which can be validated by the execution details.
Let's go back and try change the order of the join
and see the difference in the execution time of stage one.
All right, let's check what table was used in stage two.
So in stage two over here,
it's written data from table line item.
And in stage three, it's written data from table part.
So here, now stage two is filling 7.8 gigs of data,
and stage three is filing 267 megabytes of data.
So this is like the correct order of join now,
which means that stage two,
which is reading table from line item
is on the left hand side of the join,
and stage three, which is reading part table
is on right hand side of the join.
And as you can see,
the execution time of stage one is 28 minutes.
Whereas the execution time in the previous query was,
let's go back and check.
Yeah, it was 40 minutes for stage one.
So the correct order of the join
helps in reducing the time spent in that operation
and the execution datas will help you optimize that.
Now let's take example of a query
where a number of partitions is being bottlenecked.
In this table, call one is the partition field,
which has around 100,000 partitions.
This query took around 18 seconds to run.
As you can see,
the query has filtered on call one between one and 100.
Now let's check the query stats.
Here, the time spent in planning phase is seven seconds,
which comprises of 94% of the query execution time.
In order to reduce the time spent
in planning phase that is returning metadata
from Glue catalog,
there are a couple of ways to optimize it.
You can use partition indexing on the Glue table,
which will push down the filter
which is between one and 100 to the catalog,
and it will only retrieve the metadata
ranging from 1 to 100.
I've already run an example query on this.
So here it's, again, similar table,
where call one is the partition field
and the filter is also the same.
It is around 100,000 partitions.
As you can see,
the time spent in planning phase was 565 milliseconds,
which has reduced drastically from the previous table
where the partition indexing was not relevant.
You can also use partition projection
to reduce time spent in planning phase.
This way, the query stats help you optimize
the partitioning used for your table and query.
In this example,
let's examine the GROUP BY clause behavior for this query.
As you can see,
we are using four columns in the GROUP BY clause.
I already ran the query once
and it took around 35 seconds to run.
Let's see the execution details.
From these execution details,
we see that the stage two spent maximum time in executing.
Let's see what's going on.
So this stage has ScanProject operator
which fed the output to the aggregate operator.
This aggregate is being performed
as part of the GROUP BY clause on the four columns,
which we can see over here.
Now let's say we want to reduce the runtime
of this query from 35 seconds.
So what can we do?
The GROUP BY operator distributes rows
based on the GROUP BY columns to worker nodes,
which hold the values in memory.
As the rows are being ingested,
the GROUP BY columns are looked up in memory,
and the values are compared.
So in order to optimize the performance of GROUP BY,
you can limit the number of columns
within the SELECT and GROUP BY clause
to reduce the amount of memory required to store,
because rows are held in memory
and aggregated for the GROUP BY clause.
Let's see an example where the number of GROUP BY clause
have been reduced to two compared to four.
As you can see here,
we are using only two columns in the GROUP BY clause,
and this query took around seven seconds to run.
Let's check the execution details.
As you can see,
the stage two ran in one hour and 18 minutes,
where the GROUP BY clause had only two columns,
like seen here.
Whereas the query where the number of columns
in the GROUP BY clause were four,
the stage ran in two hours and 20 minutes.
And the output data was around 38 gigabytes.
Whereas for the query that only had two columns,
the output data was four gigabytes.
So overall, this query ran faster
compared to the query with four columns
in the GROUP BY clause.
This way, the execution datas
will help you optimize different operators
used in your queries.
To help you get started, here are some resources.
The links for these resources
are in the description panel of the video.
The What's New post and Athena documentation
have more details on the features we discussed today.
The blog on Top Ten Performance Tuning Tips
is a good reference for optimizing the Athena queries.
Thanks for watching this video,
and your continued use of Athena.
If you have questions or feedback to share,
please reach out to us at the displayed email address.
Thank you.