when I run

dt.optimize().executeCompaction()

on a directory with 2411 parquet files (average file size around 150mb) with

OPTIMIZE_TARGET_SIZE

at 268mb, there were only 164 added files with file sizes much closer to 268mb. The other 2K+ files were untouched and remain at ~150mb. Any clues why the vast majority of files could not be compacted? Is this a common experience?

I suspect because 150mb is close enough to the target size, and combining any two makes the new file decently larger than the target. The files that were compacted were probably a bit smaller, so combining 2-3 hit the target size. I suspect if you set the target size to 512-1024MB there would be a substantial difference. Any reason 268MB was selected?

Thanks for the tip re target size! 268mb is the default (at least in Databricks). If you have any suggestions for an improved target size let me know. The total table size is 345 GB, I'm not using hive-style partitioning (i.e. subdirectories): rather all parquet files are in 1 directory. This is time series data in an incremental table - new data is appended daily. I'm running this now to see if it'll improve query times

dt.optimize().executeZOrderBy("TimeStamp").executeCompaction()

That should write out files around 268MB, because the z-order is a global reorder, so all the data will get comingled and rewritten.

Noted - thanks 🙂 btw I am a bit confused by the Databricks recommendation to not partition tables that are less than 1TB. If so, I should

df.coalesce(1)

my entire table? Also, why would the default

OPTIMIZE_TARGET_SIZE

be more than 3 orders of magnitude smaller 🤔

oh, because ideal storage layout of your data involves two things: each parquet file being a size that results in good throughput from the storage system (smaller number of bigger files, but still a good number of files to read each from an executor to get good parallelism) and ordering on columns that have predicates applied to them so the min/max stats on the columns per file have tighter ranges. This allows selective queries to skip reading the files that cannot have answers based on the min/max being out of range. It also allows good throughput from storage when a query involves most or all the table, while at the same time reducing the size of the table log file. The data that Databricks has been able to evaluate shows that partitioning on tables under 1TB tends to create too many smaller files, and that either z-ordering or doing ingestion time clustering results in more optimal file layoutwrt size, number, and useful metadata stats.

I will run some tests with different partition sizes to see which gives the best performance. Would z-ordering still make sense on a very high cardinality column like timestamp, or do you think for best performance I should lower the granularity by creating a date column from timestamp?

Z-ordering is specifically for high cardinality columns. Don’t partition.

I ran a few experiments to test the sensitivity of file sizes on query times (always querying the same 3 days of data), on the same 354GB table with time series data from 2018-2023: • original dataset (file sizes ~250mb): query time = 2 seconds •

dt.optimize().executeZOrderBy("TimeStamp")

with

targetFileSize=1gb

: query time = 15 seconds •

dt.optimize().executeZOrderBy("TimeStamp")

with

targetFileSize=100mb

: query time = 1.2 seconds • just for fun:

repartition(1326)

to shuffle all data: query time= 20 mins 🙃 The trend looks like the smaller the parquet file, the faster the query... I'm not so convinced that having a single parquet file of 354GB would be a good idea 😄

Hahhah Nobody said anything about a single 345GB parquet. Yeah so your test is totally skewed to what works best for the type of query you are doing and on a single instance cluster. If your query is selective based on an ordered column, then small files will result in processing the least data on the single node. But when randomize the data into many files, you get what it takes to do a full table scan.

"Hahhah Nobody said anything about a single 345GB parquet." Oh I think I misunderstood the recommendation in the docs. "Not to partition tables less than 1TB", means not having the hive-style

.partionBy

subdirectory style of storage on disc.. and it doesn't mean all data in 1 parquet file...

Ha, yes exactly—no hive-style partitioning

That's exactly why I was so dumbstruck by that part of the docs... let's hope other users are not as foolish as me

Partitioning means separating data files by value of a specific column (tuple), so for instance if there is a date column that is a partition, all data files will only include data for the same date values. In delta the partition value, say a specific date in this example, is also stored in the log for each file. When your query references a specific date, the log is processed and only the data files with that matching date value as partition would be read. Partitions are generally used on lower cardinality columns, otherwise the partitions tend to fragment the data in each write into small files, and even with compaction each partition still has only small files in it. If you only ever query a partition at a time (again single date in this example), then this will give you fast result because only a tiny bit of data is read. However, if you have to query a large range of partitions the performance will be very bad due to suboptimal throughput to the storage system having to read many small files.

That's a clear explanation. I'm soaking up every word! Thank you for teaching me!

No problem!

Even for a query spanning a window of 1 year, the larger file sizes are slower. Here I am comparing the same query between two storage structures for the same table (345 GB)

Did you run this on a single node cluster again?

Yes. 1 node as in the screenshot I shared yesterday. I get the hint. Thanks!

My point is not that using a single node cluster is "wrong," but rather the results are only valid if you will only be using single node clusters of similar resource type to query the table. You are likely to get different results on say a four node i3.4xlarge cluster for example. This is the challenge with benchmarking as a practice--both the people doing the benchmarking and the consumers of the benchmark have to understand how the conditions of the measurements will affect them and what might be extrapolated safely from the results.

• in the 1st table (hive style), I don't see any indication of compaction or z-ordering from looking at the delta log .json files. I just see "add"s:

{"add":{"path":"SiteNumber=BE01/Year=2023/Month=8/Day=1/part-00004-2dfc8f9a-1f94-4d87-881a-5f970cb00451.c000.snappy.parquet"

. The files are simply written into daily bins. • In the 2nd, I am z-ordering on TimeStamp, with targetFileSize=268mb.