Enable parallel backfill in run.py

[pengyu-hou]

Summary

This PR will add a new param parallelism to run.py to use with start-ds and end-ds. The purpose of this param is to allow users submit multiple spark jobs to backfill data in parallel.

Why / Goal

This enables users a better backfilling experience

Test Plan

• Added Unit Tests
• Covered by existing CI
• Integration tested
• tested on GW: SPARK_VERSION=3.1.1 python3 ./scripts/run.py --mode=backfill --conf=production/staging_queries/team/testing_config.v1 --start-ds 2023-12-08 --end-ds 2023-12-31 --parallelism 6 | tee ~/test/staging_query.log

## Checklist - [ ] Documentation update

Reviewers

@airbnb/zipline-maintainers

[hzding621]

curious why we run the first command by itself?

[pengyu-hou]

Actually, it is not necessary. I made some simplification.

[cristianfr]

As I understand it.
Create --start-ds --end-ds + parallelism:
Divides the job into parallel date ranges (as opposed to step-days which is sequential date ranges).

This can an effect with failing jobs though right? It may be hard to recover from one range that fails within the sequence of jobs.

Another concern, some of our jobs in the spark level have hole filling logic, iirc. So the newer jobs may (in parallel) may find the hole and try to fill it (all at the same time).

So what I'm thinking is we have a few systems that look into the past data:

• Hole filling logic
• Define compute based on existing data. (Partitions to fill I think is the name).
• This would parallelism at the run.py level.

What's the advantage of doing it at the run.py level vs having the spark job parallelize jobs itself (which would prevent the conflict between multiple spark jobs (unaware of each other) trying to fill the same date range).

[pengyu-hou]

As I understand it. Create --start-ds --end-ds + parallelism: Divides the job into parallel date ranges (as opposed to step-days which is sequential date ranges).

This can an effect with failing jobs though right? It may be hard to recover from one range that fails within the sequence of jobs.

Another concern, some of our jobs in the spark level have hole filling logic, iirc. So the newer jobs may (in parallel) may find the hole and try to fill it (all at the same time).

So what I'm thinking is we have a few systems that look into the past data:

• Hole filling logic
• Define compute based on existing data. (Partitions to fill I think is the name).
• This would parallelism at the run.py level.

What's the advantage of doing it at the run.py level vs having the spark job parallelize jobs itself (which would prevent the conflict between multiple spark jobs (unaware of each other) trying to fill the same date range).

Had a chat with @cristianfr and @nikhilsimha offline that this will be our short term approach for backfill performance. We will continue to work on a long term approach on the aggregation core to further improve the performance.