How many files are output by a Foundry Transform in various combinations of repartition, hive partitioning, and bucketing?

For hive partitioning + bucketing, the # of output files is not constant and will depend on the actual data of the input partition.To clarify, let's say df is 200 partitions, not 200 files. Output files scale with # of input partitions, not # of files. 200 files could be misleading as that could be 1 partition to 1000's of partitions.

First a few ones to check my understanding:

df.repartition(100) = 100 output files of the same size

df.repartition('colA') = 10 output files of different sizes, since each file will contain all rows for 1 value of colA

df.repartition('colB') = 1000 output files

df.repartition(50, 'colA') = 50 output files

df.repartition(50, 'colB') = 50 output files

Hive partitions:

output.write_dataframe(df, partition_cols=['colA']) = upper bound of 2,000 output files (200 input partitions * max 10 values per partition)

output.write_dataframe(df, partition_cols=['colB']) = max 200,000 output files (200 * 1000 values per partition)

output.write_dataframe(df, partition_cols=['colA', 'colB']) = max 2,000,000 output files (200 partitions * 10 values * 1000)

output.write_dataframe(df.repartition('colA'), partition_cols=['colA']) = 10 output files of different sizes (1 file in each hive partition)

Bucketing:

output.write_dataframe(df, bucket_cols=[‘colB’], bucket_count=100) = max 20,000 files (200 partitions * max 100 buckets per partition)

output.write_dataframe(df, bucket_cols=[‘colA’], bucket_count=10) = max 2,000 files (200 partitions * max 10 buckets per partition)

output.write_dataframe(df.repartition(‘colA’), bucket_cols=[‘colA’], bucket_count=10) = exactly 10 files (repartitioned dataset makes 10 input partitions, each partition outputs to only 1 bucket)

All together now:

output.write_dataframe(df, partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200) = I could be wrong on this, but I believe it's max of 400,000 output files (200 input partitions * 10 colA partitions * 200 colB buckets)

output.write_dataframe(df.repartition(‘colA’, ‘colB’), partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200) = I believe this is exactly 10,000 output files (repartition colA,colB = 10,000 partitions, each partition contains exactly 1 colA and 1 bucket of colB)

Background

The key to being able to reason about output file counts is understanding at which level each concept applies.

Repartition (df.repartition(N, 'colA', 'colB')) creates a new spark stage with the data shuffled as requested, into the specified number of shuffle partitions. This will change the number of tasks in the following stage, as well as the data layout in those tasks.

Hive partitioning (partition_cols=['colA', 'colB']) and bucketing (bucket_cols/bucket_count) only have any effect within the scope of the final stage's tasks, and effect how the task writes its data into files on disk.

In particular, each final stage task will write one file per hive-partition/bucket combination present in its data. Combinations not present in that task will not write an empty file if you're using hive-partitioning or bucketing.

Note: if not using hive-partitioning or bucketing, each task will write out exactly one file, even if that file is empty.

So in general you always want to make sure you repartition your data before writing to make sure the data layout matches your hive-partitioning/bucketing settings (i.e. each hive-partition/bucket combination is not split between multiple tasks), otherwise you could end up writing huge numbers of files.

Your examples

I think there is some misunderstanding floating around, so let's go through these one by one.

First a few ones to check my understanding:

df.repartition(100) = 100 output files of the same size

Yes - the data will be randomly, evenly shuffled into 100 partitions, causing 100 tasks, each of which will write exactly one file.

df.repartition('colA') = 10 output files of different sizes, since each file will contain all rows for 1 value of colA

No - the number of partitions to shuffle into is unspecified, so it will default to 200. So you'll have 200 tasks, at most 10 of which will contain any data (could be fewer due to hash collisions), so you will end up with 190 empty files, and 10 with data. *Note: with AQE in spark 3, spark may decide to coalesce the 200 partitions into fewer when it realizes most of them are very small. I don't know the exact logic there, so technically the answer is actually "200 or fewer, only 10 will contain data".

df.repartition('colB') = 1000 output files No - Similar to above, the data will be shuffled into 200 partitions. However in this case they will (likely) all contain data, so you will get 200 roughly-equally sized files.

Note: due to hash collisions, files may be larger or smaller depending on how many values of colB happened to land in each partition.

df.repartition(50, 'colA') = 50 output files?

Yes - Similar to before, except now we've overridden the partition count from 200 to 50. So 10 files with data, 40 empty. (or fewer because of AQE)

df.repartition(50, 'colB') = 50 output files, so some files will contain more than one value of colB?

Yes - Same as before, we'll get 50 files of slightly varying sizes depending on how the hashes of the colB values work out.

Hive partitions:

(I think the below examples are written assuming df is in 100 partitions to start rather than 200 as specified, so I'm going to go with that)

output.write_dataframe(df, partition_cols=['colA']) = 1,000 output files (because I get potentially 100 files in each of the 10 hive partitions 10)

Yes - You'll have 100 tasks, each of which will write one file for each colA value they see. So up to 1,000 files in the case the data is randomly distributed.

output.write_dataframe(df, partition_cols=['colB']) = 10,000 output files

No - Missing a 0 here. 100 tasks, each of which could write as many as 1,000 files (one for each colB value), for a total of up to 100,000 files.

output.write_dataframe(df, partition_cols=['colA', 'colB']) = 100,000 output files

No - 100 tasks, each of which will write one file for each combination of partition cols it sees. There are 10,000 such combinations, so this could write as many as 100 * 10,000 = 1,000,000 files!

output.write_dataframe(df.repartition('colA'), partition_cols=['colA']) = 10 output files of different sizes (1 file in each hive partition)

Yes - The repartition will shuffle our data into 200 tasks, but only 10 will contain data. Each will contain exactly one value of colA, so will write exactly one file. The other 190 tasks will write no files. So 10 files exactly.

Bucketing:

Again, assuming 100 partitions for df, not 200

output.write_dataframe(df, bucket_cols=[‘colB’], bucket_count=100) = 100 output files? In an experiment, this did not seem to be the case

No - Since we haven't laid out the data carefully, we have 100 tasks with (maybe) randomly distributed data. Each task will write one file per bucket it sees. So this could write up to 100 * 100 = 10,000 files!

output.write_dataframe(df, bucket_cols=[‘colA’], bucket_count=10) = 10 output files?

No - Similar to above, 100 tasks, each could write up to 10 files. So worst-case is 1,000 files here.

output.write_dataframe(df.repartition(‘colA’), bucket_cols=[‘colA’], bucket_count=10) = ???

Now we're adjusting the data layout before writing, we'll have 200 tasks, at most 10 of which will contain any data. Each value of colA will exist in only one task.

Each task will write one file per bucket it sees. So we should get at most 10 files here.

Note: Due to hash collisions, one or more buckets might be empty, so we might not get exactly 10.

All together now:

Again, assuming 100 partitions for df, not 200

output.write_dataframe(df, partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200) = ???

100 tasks. 10 hive-partitions. 200 buckets. Worst case is each task writes one file per hive-partition/bucket combination. i.e. 100 * 10 * 200 = 200,000 files.

output.write_dataframe(df.repartition(‘colA’, ‘colB’), partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200) = ??? -- Is this the command that I want to use in the end? And anything downstream would first filter on colA to take advantage of the hive partitioning, then join on colB to take advantage of the bucketing?

This one is sneaky. We have 200 tasks and the data is shuffled carefully so each colA/colB combination is in just one task. So everything seems good.

BUT each bucket contains multiple values of colB, and we have done nothing to make sure that an entire bucket is localized to one spark task.

So at worst, we could get one file per value of colB, per hive partition (colA value). i.e. 10 * 1,000 = 10,000 files.

Given our particular parameters, we can do slightly better by just focusing on getting the buckets laid out optimally:

output.write_dataframe(df.repartition(200, ‘colB’), partition_cols=[‘colA’], bucket_cols=[‘colB’], bucket_count=200)

Now we're making sure that colB is shuffled exactly how it will be bucketed, so each task will contain exactly one bucket.

Then we'll get one file for each colA value in the task (likely 10 since colA is randomly shuffled), so at most 200 * 10 = 2,000 files.

This is the best we can do, assuming colA and colB are not correlated.

Conclusion

There's no one-size fits all approach to controlling file sizes.

Generally you want to make sure you shuffle your data so it's laid out in accordance with the hive-partition/bucketing strategy you're applying before writing.

However the specifics of what to do may vary in each case depending on your exact parameters.

The most important thing is to understand how these 3 concepts interact (as described in "Background" above), so you can reason about what will happen from first principals.