0.10.0 RFC DynamicOutput

[alangenfeld]

In the first step in supporting runtime dynamic dag structures in dagster (discussed in issue #462), we are shooting to add fan-out / foreach / mapping capabilities in an experimental state for 0.10.0. We expect it to look something like this:

@solid(output_defs=[DynamicOutputDefinition()])
def X(_):
    for i in range(3):
        yield DynamicOutput(value=i, mappable_key=str(i))

@pipeline
def example():
    y_out = Y()
    items = X()

    # exact DSL syntax up for discussion below
    def _downstream(entry):
        a_1, a_2 = A(entry)
        D(B(a_1, y_out), C(a_2))

    items.map(_downstream)

[schrockn]

I'm concerned about the use of the for loop here, since now the for loop can mean two different things in this context.

@pipeline
def another_example():
  for i in range(0, 4):
    get_solid(name=f"solid_{i}")

In this case it would produce 4 static solids in parallel, rather than use dynamic orchestration

[schrockn]

Alex is proposing an alternative semantic where if the for loop is over a solid with a mappable output, it instead does dynamic orchestration semantics.

[alangenfeld]

Ya the for loop approach for unpacking is definitely a bit cute, @catherinewu raised similar concerns about it being confusing. I will edit the discussion summary to make it clear this aspect is still up for discussion.

if the for loop is over a solid with a mappable output, it instead does dynamic orchestration semantics.

This particular framing doesn't hit as hard with me - what you are iterating over is just as important as the for keyword, so having these two types of loops potentially coexist seems acceptable to me.

The argument that resonates with me is it just making the function based DSL even more tricky than it already is, especially to someone who doesn't know python that well.

That said, I don't love the alternatives I have come up with either. The assumption I am operating from is that we want to have some unpacking step to make it clear that mapping will occur.

# unpack on the output
@pipeline
def example():
    y_out = Y()
    items = X()
    entry = items.forEach() # / items.mapped()
    a_1, a_2 = A(entry)
    D(B(a_1, y_out), C(a_2))

# unpack when used
@pipeline
def example():
    y_out = Y()
    items = X()
    a_1, a_2 = A.forEach(items) # / A.mapped(items)
    D(B(a_1, y_out), C(a_2))

# unpack as function
# could use lambda when 1 line
def _map_A_item(item):
    a_1, a_2 = A(entry)
    D(B(a_1, y_out), C(a_2))

@pipeline
def example():
    y_out = Y()
    items = X()
    items.forEach(_map_A_item) # / items.map(_map_A_item)

# context manager
@pipeline
def example():
    y_out = Y()
    items = X()
    with items.forEach() as entry: / # items.mapped()
        a_1, a_2 = A(entry)
        D(B(a_1, y_out), C(a_2))

Do you all like any of these ^ or have a different approach that would be good?

[sryza]

Driving by with a half-formed suggestion: one lens to help decide could be to think about how explainable the nouns are in each option. E.g. "in example X, a_1 is a dynamically-sized collection of logical solid outputs."

Some other thoughts:

• I am biased against A.forEach. I think users are not used to calling methods on functions or things that behave like functions.
• I think we should prefer map to forEach, because map implies a return value, while forEach does not.
• A question about the for loop approach - how does the user get back to a collection? In regular programming, if I have lots of stuff to do, per item, I can break it up:

my_collection = ...

my_processed_collection = []
for item in my_collection:
    my_processed_collection.append(a(b(c(item)))

my_further_processed_collection = []
for item in my_processed_collection:
    my_further_processed.append(d(e(f(item)))

...

[alangenfeld]

A question about the for loop approach - how does the user get back to a collection? In regular programming, if I have lots of stuff to do, per item, I can break it up

I believe that should be fine, given how I expect the __iter__ approach to be implemented. Walking through with some inline comments:

# assume my_collection is result of invoking solid with mappable output
my_collection = ... 

my_processed_collection = []
# __iter__ on the special mapping output handle unpacks it and returns 1 item
# a handle that establishes the dependency as downstream of mapping
for item in my_collection:
    my_processed_collection.append(a(b(c(item)))
    # my_processed_collection now has 1 entry, the output handle from a, downstream of mapping


my_further_processed_collection = []
# iterate over the one item in the collection
for item in my_processed_collection:
    my_further_processed_collection.append(d(e(f(item)))
    # my_further_processed_collection now has 1 entry:
    # the output handle from d, downstream from a, downstream from mapping

# This is out of scope for what we expect to release for 0.10.0, but
# in the same way we treat lists as the way to do fan-in in the DSL, I expect we
# will treat fan-in of outputs downstream of fan-out as collecting fan-in.
collect(my_further_processed_collection)

[alangenfeld]

edited OP to incorporate feedback on for loop -> map and mappable -> dynamic rename

[alangenfeld]

This shipped in 0.10.0, with front-end support to come in subsequent releases.