How to do a bucle in a Dagster pipeline?

[ericbellet]

Hello! I just receiving two dates, and for each day between this 2 dates I need to run a pipeline, how would you do it? Basically I need to run the pipeline for each date.
I wrote the following code:

from dagster import (
    DagsterType,
    InputDefinition,
    OutputDefinition,
    String,
    execute_pipeline,
    pipeline,
    solid,
)
from datetime import datetime, timedelta


@solid
def setDatesToProcess(context, startDt, endDt):
    startDt = datetime.strptime(startDt, "%Y-%m-%d")
    endDt = datetime.strptime(endDt, "%Y-%m-%d")
    dates = [startDt + timedelta(days=x) for x in range((endDt - startDt).days + 1)]
    return dates

@solid
def obtainDataFromKafka(context, date):
    context.log.info(f"Extracting data from {date}")
    return ["Example", "Data"]

@solid
def saveFileS3(context, data, date):
    path = f"s3:mybucket/myfolder/{date}"
    context.log.info(f"Saving {data} in the path {path}")
    return path

@solid
def copys3FileToTableRedshift(context, path):
    context.log.info(f"Copying to Redshift using S3 file {path}")

@pipeline
def my_pipeline():
    dates = setDatesToProcess()
    for date in dates:
        copys3FileToTableRedshift(saveFileS3(obtainDataFromKafka(date)))
run_config = {
    "solids": {
        "setDatesToProcess": {
            "inputs": {
                "startDt": {"value": "2020-10-05"},
                "endDt": {"value": "2020-10-10"}
            }
        }
    }
}

if __name__ == "__main__":
    execute_pipeline(my_pipeline, run_config=run_config)

Another question, How would you pass the variable date in each solid?

[sryza]

Hey @ericbellet. The capability you're describing is something that we're working on, but haven't completed implementing. The way Dagster currently works, the set of solids is fixed inside a pipeline definition. I.e., after the pipeline definition has been created, there's no way for a solid to say "launch a downstream solid for every output that I return".

If you're willing to condense those parallel task into a single task, you could do something like what was suggested here: #3211.

If you're able to accept those time parameters outside of Dagster config, you could do something like this:

from dagster import (
    DagsterType,
    InputDefinition,
    OutputDefinition,
    String,
    execute_pipeline,
    pipeline,
    solid,
)
from datetime import datetime, timedelta


@solid
def obtainDataFromKafka(context, date):
    context.log.info(f"Extracting data from {date}")
    return ["Example", "Data"]

@solid
def saveFileS3(context, data, date):
    path = f"s3:mybucket/myfolder/{date}"
    context.log.info(f"Saving {data} in the path {path}")
    return path

@solid
def copys3FileToTableRedshift(context, path):
    context.log.info(f"Copying to Redshift using S3 file {path}")

def make_pipeline(startDt, endDt):
    @pipeline
    def my_pipeline():
        startDt = datetime.strptime(startDt, "%Y-%m-%d")
        endDt = datetime.strptime(endDt, "%Y-%m-%d")
        dates = [startDt + timedelta(days=x) for x in range((endDt - startDt).days + 1)]
        for date in dates:
            copys3FileToTableRedshift(saveFileS3(obtainDataFromKafka(date)))

if __name__ == "__main__":
    startDT = sys.argv[0]
    endDT = sys.argv[1]
    execute_pipeline(make_pipeline(startDT, endDT))

[sryza]

For your last question: #3213