Hi again guys, I've got a question on pipelines to...
# announcements
Hi again guys, I've got a question on pipelines today.
I've got a pipeline that looks like this:
Copy code
def crawlers_pipeline():
    running_jobs = run_crawlers()
    b = wait_for_jobs(running_jobs)
    c = get_results(b)
run_crawlers is a solid that starts several crawler jobs using an API, and returns a dict containing objects used in the following steps to access job status and results
My problem is that some of the crawlers take a lot longer to run than the others, and wait_for_jobs has to wait for them all to finish
I'm wondering if I could use solid aliases to "branch out" out of run_crawlers and execute the following solids once for each of the crawlers
something like:
Copy code
def comparativos_pipeline():
    running_jobs = run_crawlers()
    for crawler in Crawlers:
        a_wait_for_jobs = wait_for_jobs.alias("wait_for_{}".format(crawler))
        a_get_results = get_results.alias("get_{}_results".format(crawler))
        a_write_results_to_s3 = write_results_to_s3.alias("write_{}_results".format(crawler))
        b = a_wait_for_jobs(running_jobs[crawler])
        c = a_get_results(b)
Is this possible? I'm wondering how to yield and "unpack" the results from run_crawlers
ya we dont have generic map/reduce support yet but you can set up a fixed width parallelism set up - here is a toy example https://github.com/dagster-io/dagster/blob/master/examples/dagster_examples/toys/sleepy.py
do you know how many jobs are going to happen at pipeline creation time? you could also just dynamically generate a pipeline that fans out for every expected job
Ah yes, that should work perfectly! Thank you!
The jobs are known at creation time and yes, I'm attempting to generate it dynamically
With maybe the possibility of turning jobs on and off via config
things to look at if you want to do more generation: * creating
for each named job and yielding
Output(job, 'job_name')
for each job * setting
to be able to turn off a job by not yielding the
(toggled via config) * in the composition function - we return a namedtuple for multiple outputs - you will interact with that to wire up the named job output to its aliased downstream dag
Copy code
def handle_job(job_spec: CrawlerJob):
    b = wait_for_jobs(job_spec)
    c = get_results(b)

def generate_crawlers_pipeline():
    job_specs = get_jobs() # dict[str, job]
    outputs_defs = [
        OutputDefinition(CrawlerJob, name, is_required=False)
        for name in job_specs.keys()

        config={'skip': Field([str], is_required=False, default_value=[])}
    def run_crawlers(context):
        for name, job in job_specs:
            if name not in context.solid_config['skip']:
                job = do_work(job)
                yield Output(job, name)

    def crawlers_pipeline():
        outputs = run_crawlers()
        # convert namedtuple to dict for iteration
        # might be better way to do this
        for name, handle for outputs._asdict():

    return crawlers_pipeline