Suggestion: Simplify Sequential and/or Concurrent job scheduling

[clayheaton]

It is a pain in the butt to set up a series of jobs that should execute sequentially (or even concurrently). To set them up sequentially, it's easiest to have Job 2 be attached to a Sensor that triggers off of the final step in Job 1. What that is easy enough with 2-3 jobs that have to run sequentially, it's a confusing mess when there are 50 jobs that must run sequentially.

Similarly, it's a bit of a pain to set up concurrent schedules for jobs that need to run concurrently.

I propose the addition of SequentialJobScheduleDefinition and ConcurrentJobScheduleDefinition:

job1 = define_asset_job(
    name="Job1",
    description="The first job",
    selection=AssetSelection.groups(JOB1_GROUP)
)

job2 = define_asset_job(
    name="Job2",
    description="The second job",
    selection=AssetSelection.groups(JOB2_GROUP)
)

# Etc for any number of jobs

# Here the jobs execute in sequence. job2 starts when job1 is complete, etc.
sequential_job_schedule = SequentialJobScheduleDefinition(
    job_sequence=[
        job1,
        job2,
        # ...
    ],
    cron_schedule="0 14 * * Tue",
    execution_timezone="America/New_York",
    default_status=DefaultScheduleStatus.STOPPED,
)

defs = Definitions(
    assets=...,
    resources={...},
    jobs=[
        job1,
        job2,
        ...
    ],
    sensors=[...],
    schedules=[
        sequential_job_schedule,
        ...
    ],
)

Similarly, for concurrent jobs:

job1 = define_asset_job(
    name="Job1",
    description="The first job",
    selection=AssetSelection.groups(JOB1_GROUP)
)

job2 = define_asset_job(
    name="Job2",
    description="The second job",
    selection=AssetSelection.groups(JOB2_GROUP)
)

# Etc for any number of jobs

# Here the jobs execute concurrently
concurrent_job_schedule = ConcurrentJobScheduleDefinition(
    jobs=[
        job1,
        job2,
        # ...
    ],
    maximum_concurrent_jobs=3, # The maximum jobs that run concurrently when this executes
    cron_schedule="0 14 * * Tue",
    execution_timezone="America/New_York",
    default_status=DefaultScheduleStatus.STOPPED,
)

defs = Definitions(
    assets=...,
    resources={...},
    jobs=[
        job1,
        job2,
        ...
    ],
    sensors=[...],
    schedules=[
        concurrent_job_schedule,
        ...
    ],
)