When working on the Refine refactoring T356762 , I noticed that we might do better considering the parallelization of the process and the desired number and size of files in the output.

1/ Parallelization

With the refactoring, we get three sizes of spark job:

• small: local (to the Skein app) with one worker
• medium: Spark cluster with a limit of 4 workers of 8GB
• large: Spark cluster with a limit of 32 workers of 8GB

On a large one with a partition input of 350MB (json.gz), we get 200 tasks matching the number of cores, which means each task is Refining less than 2MB.

In place of fixed numbers, we may control the parallelization by setting a large maximum number of executors and setting Adaptive Query Execution to true. In this case, Spark limits the size of its shuffle partitions to an HDFS block (or smaller).

With our previous example, It's going to create nine tasks. That means less overhead, smaller workers, better use of the worker, less strain on the Yarn cluster (by triggering Refine with Airflow, we are now creating as many Refine tasks as we have streams ~ 160), and auto control of the number of output.

2/ Number and size of output files

The input of Refine is Gobblin's output, which produces 1 file per topic and run. That usually means 2 or 4 files per partition, whatever the stream size. Trying to match the number of Refine outputs to the number of inputs does not make sense to me.

We can estimate the data size (as Java objects) in each Spark partition, I found AQE with a 64MB shuffle file leads to 1GB of Java objects. It also allows us to control the size of the executors.

Alternatively, it's also possible to get a total estimate of the java objects of the RDD, determine the size of each output file, and repartition the DF accordingly.

Note: Besides, as I tried to optimize the Refine output to get files the size of an HDFS block (256MB), I noticed that a 256MB compressed file led to 4GB of Java objects, which may not match our typical worker taxonomy.