All tables in Treasure Data are partitioned based on the time column. This is why queries that use TD_TIME_RANGE or similar predicates on the time column are efficient in Treasure Data. Presto can eliminate partitions that fall outside the specified time range without reading them.
User-defined partitioning (UDP) provides hash partitioning for a table on one or more columns in addition to the time column. A query that filters on the set of columns used as user-defined partitioning keys can be more efficient because Presto can skip scanning partitions that have matching values on that set of columns.
The benefits of UDP can be limited when used with more complex queries. The query optimizer might not always apply UDP in cases where it can be beneficial.
The Maximum Number of Partitioning Keys is Less Than or Equal to Three Keys
If the limit is exceeded, Presto causes the following error message:
'bucketed_on' must be less than 4 columns
Imports Other than SQL
The import method provided by Treasure Data for the following does not support UDP tables:
- Streaming Import
- Data Connector
If you try to use any of these import methods, you will get an error. As a workaround, you can use a workflow to copy data from a table that is receiving streaming imports to the UDP table.
Use Cases for UDP
Where the lookup and aggregations are based on one or more specific columns, UDP can lead to:
efficient lookup and aggregation queries
efficient join queries
UDP can add the most value when records are filtered or joined frequently by non-time attributes::
a customer's ID, first name+last name+birth date, gender, or other profile values or flags
a product's SKU number, bar code, manufacturer, or other exact-match attributes
an address's country code; city, state, or province; or postal code
Performance benefits become more significant on tables with >100M rows.
UDP can help with these Presto query types:
"Needle-in-a-Haystack" lookup on the partition key
Aggregations on the partition key
Very large joins on partition keys used in tables on both sides of the join
Basic UDP Usage
CREATE TABLE Syntax for UDP
To create a UDP table:
Use CREATE TABLE with the attributes bucketed_on to identify the bucketing keys and bucket_count for the number of buckets.
Optionally, define the max_file_size and max_time_range values.
For bucket_count the default value is 512. This should work for most use cases.
max_file_size will default to 256MB partitions
max_time_range to 1d or 24 hours for time partitioning
bucket_count = 512
Choosing Bucketing Columns for UDP
Supported TD data types for UDP partition keys include int, long, and string. These correspond to Presto data types as described in About TD Primitive Data Types.
Choose a set of one or more columns used widely to select data for analysis-- that is, one frequently used to look up results, drill down to details, or aggregate data. For example, depending on the most frequently used types, you might choose:
Country + State/Province + City
Customer-first name + last name + date of birth
Choose a column or set of columns that have high cardinality (relative to the number of buckets), and are frequently used with equality predicates. For example:
Unique values, for example, an email address or account number
Non-unique but high-cardinality columns with relatively even distribution, for example, date of birth
Checking For and Addressing Data Skew
The performance is inconsistent if the number of rows in each bucket is not roughly equal. For example, if you partition on the US zip code, urban postal codes will have more customers than rural ones.
To help determine bucket count and partition size, you can run a SQL query that identifies distinct key column combinations and counts their occurrences. For example:
If the counts across different buckets are roughly comparable, your data is not skewed.
For consistent results, choose a combination of columns where the distribution is roughly equal.
If you do decide to use partitioning keys that do not produce an even distribution, see Improving Performance with Skewed Data.
Using INSERT and INSERT OVERWRITE to Partitioned Tables
INSERT and INSERT OVERWRITE with partitioned tables work the same as with other tables. You can create an empty UDP table and then insert data into it the usual way. The resulting data is partitioned.
Partitioning an Existing Table
Tables must have partitioning specified when first created. For an existing table, you must create a copy of the table with UDP options configured and copy the rows over. To do this use a CTAS from the source table.
When partitioning an existing table:
Creating a partitioned version of a very large table is likely to take hours or days. Consult with TD support to make sure you can complete this operation.
If the source table is continuing to receive updates, you must update it further with SQL. For example:
Creating and Using UDP Tables: Examples
Create a partitioned copy of the customer table named customer_p, to speed up lookups by customer_id;
Create and populate a partitioned table customers_p to speed up lookups on "city+state" columns:
UDP Advanced Use Case Details
Choosing Bucket Count, Partition Size in Storage, and Time Ranges for Partitions
Bucket counts must be in powers of two. A higher bucket count means dividing data among many smaller partitions, which can be less efficient to scan. TD suggests starting with 512 for most cases. If you aren't sure of the best bucket count, it is safer to err on the low side.
We recommend partitioning UDP tables on one-day or multiple-day time ranges, instead of the one-hour partitions most commonly used in TD. Otherwise, you might incur higher costs and slower data access because too many small partitions have to be fetched from storage.
Aggregations on the Hash Key
Using a GROUP BY key as the bucketing key, major improvements in performance and reduction in cluster load on aggregation queries were seen. For example, you can see the UDP version of this query on a 1TB table:
used 10 Presto workers instead of 19
ran in 45 seconds instead of 2 minutes 31 seconds
processing >3x as many rows per second. The total data processed in GB was greater because the UDP version of the table occupied more storage.
Needle-in-a-Haystack Lookup on the Hash Key
The largest improvements – 5x, 10x, or more – will be on lookup or filter operations where the partition key columns are tested for equality. Only partitions in the bucket from hashing the partition keys are scanned.
For example, consider:
customersis bucketed on
contactsis bucketed on
These queries will improve:
Here UDP Presto scans only one bucket (the one that 10001 hashes to) if customer_id is the only bucketing key.
Here UDP Presto scans only the bucket that matches the hash of country_code 1 + area_code 650.
These queries will not improve:
Here UDP will not improve performance, because the predicate doesn't use '='.
Here UDP will not improve performance, because the predicate does not include both bucketing keys.
Very Large Join Operations
Very large join operations can sometimes run out of memory. Such joins can benefit from UDP. Distributed and colocated joins will use less memory, CPU, and shuffle less data among Presto workers. This may enable you to finish queries that would otherwise run out of resources. To leverage these benefits, you must:
Make sure the two tables to be joined are partitioned on the same keys
Use equijoin across all the partitioning keys
Set the following options on your join using a magic comment:
Improving Performance on Skewed Data
When processing a UDP query, Presto ordinarily creates one split of filtering work per bucket (typically 512 splits, for 512 buckets). But if data is not evenly distributed, filtering on skewed bucket could make performance worse -- one Presto worker node will handle the filtering of that skewed set of partitions, and the whole query lags.
To enable higher scan parallelism you can use:
When set to true, multiple splits are used to scan the files in a bucket in parallel, increasing performance. The tradeoff is that colocated join is always disabled when distributed_bucket is true. As a result, some operations such as GROUP BY will require shuffling and more memory during execution.
This query hint is most effective with needle-in-a-haystack queries. Even if these queries perform well with the query hint, test performance with and without the query hint in other use cases on those tables to find the best performance tradeoffs.