Hi wave We are using Delta Lake 1 2 on EMR and I m trying to

Question

Hi wave We are using Delta Lake 1 2 on EMR and I m trying to understand why our largest tables all end up with exactly 1 000 files in S3 and this does not change as we merge in more data They are of varying sizes e g 1 1 TB 7B records 330GB 1 5B records 53 GB 180M records but all have 1 000 files even though we have not explicitly repartitioned nor set shuffle partitions I know delta now does repartition before merge by default could that have an impact It looks as if there is a max setting at play somewhere

Scott Sandre (Delta Lake) · Answer

This seems odd Have you double checked this Where did you see the `1000` number Is it paging results

Scott Sandre (Delta Lake) · Answer

gt I know delta now does repartition before merge by default could that have an impact No If the table has 1000 parquet files at time T and then at time T+1 you perform a merge then delta will have to re write some parquet files so surely there will be more than 1000 parquet files

Matt Richards · Answer

The Spark UI query plan shows `number of files read 1000`

Scott Sandre (Delta Lake) · Answer

that s different from `the table contains exactly 1000 files` btw

Scott Sandre (Delta Lake) · Answer

what query are you running

Scott Sandre (Delta Lake) · Answer

what do your merges look like

Matt Richards · Answer

I can t paste a screenshot unfortunately The merges are building a fact table from upstream event table and joining to dimensions so vary in complexity The upstream source DF has 1d of new updated records that merge into the main fact target

Matt Richards · Answer

the files read is coming from a Scan parquet step that has no filters that doesn t tell you the number of actual parquet files

Scott Sandre (Delta Lake) · Answer

gt that doesn t tell you the number of actual parquet files I m just trying to highlight that if you insert 1000 files then remove 1000 files then insert 1000 files you will have 2000 files in your table because we tombstone files not immediately remove them So if you insert 1000 files then do a merge we may logically remove 100 files then insert 100 files and you will have 1100 files in your table

Scott Sandre (Delta Lake) · Answer

but this isn t helpful it seems to you debugging this problem slightly smiling face

Scott Sandre (Delta Lake) · Answer

if you can recreate this using sample data sample queries that would probably be the next best step for me to help you here

Matt Richards · Answer

slightly smiling face

Matt Richards · Answer

Unfortunately we don t have a non prod environment that gets close to the same volumes The delta metrics show 1000 files added and 1000 files removed The merge uses the event key I thought that phase 1 of the merge was an inner join of source to target to identify files that will be modified We have 1 846 155 rows in source merging to 7 522 723 519 so am puzzled why ends up rewriting the entire table with no change in file count I can try to replicate it in dev with scaled volumes

chris fish · Answer

chris fish · Answer

`describe detail` tells you the size and number of files in each tables current snapshot

Matt Richards · Answer

hmm just checked a query plan from a dev integration test that is using just a smaller number of records The merge there shows an exchange for the source side that has 15 shuffle records data size of 9 6KiB and number of partitions 1 000 thinking face

chris fish · Answer

sorry so is this the number of shuffle partitions somewhere on your cluster

chris fish · Answer

can you paste the spark ui screenshot here

Matt Richards · Answer

image png

Matt Richards · Answer

As far as I can see there s nothing setting shuffle partitions but it does look like that s the obvious explanation

Matt Richards · Answer

In prod the equivalent looks like

Matt Richards · Answer

Nothing set in spark properties for shuffle partitions according to UI Environment

chris fish · Answer

weird that is very consistently being set to 1000 shuffle partitions

chris fish · Answer

you can set the shuffle partitions during runtime it doesn t necessarily show up in the sparkenv UI

Matt Richards · Answer

< chris fish> < Scott Sandre Delta Lake > it looks like EMR has a default setting out of the box of `spark sql adaptive coalescePartitions initialPartitionNum=1000`

Matt Richards · Answer

I noticed that in one query plan there are a couple of shuffles that get partitions = 200 and then later it switches to 1000 with a subsequent AQE coalesce Is it the case that AQE may not be active at a particular point in which case we end up with the default shuffle partitions Do you have any recommendations Leave that default but explicitly set shuffle partitions when we know we need it or always be explicit

chris fish · Answer

ah yeah so that setting only `coalesces` partitions down so when you are reading in a lot of data there might be 10 000 tasks reading in the data and then it will coalesce down to 1000 tasks during a shuffle but if the tasks never goes above 1000 then that AQE feature will never kick in

Matt Richards · Answer

any thoughts on the best approach here

chris fish · Answer

it sounds like in EMR you still have to manually control the number of partitions here

chris fish · Answer

you could definitely reach out to an EMR specialist and ask them why AQE isn t performing better

Matt Richards · Answer

thanks I ve asked our AWS SA to connect us with a specialist on their side