Heya! I’m having some fun with dagster (and I really like it so far tbh 🤩), and was wondering about a few things here, thought you’d might be able to help: 1. Is there a pretty way to unpack DynamicOutput into multiple arguments of a mapped op?

@op(out=DynamicOut())
def my_op():
    yield DynamicOutput(
        value={'a': 1, 'b': 2},
        key=...
    )

@op
def another_op(a, b):
    return a + b

@job
def test():
    my_op().map(another_op)

2. Also when mapping tasks results, is there a way to pass in extra arguments?

@op
def another_op(a, b):
    return a + b

@job
def test():
    mynumber = get_number()
    my_op_2().map(another_op, mynumber)

Thanks!

Hi Sa’ar! For 1: I don’t think there is special unpacking syntax. You should be able to implement a pass-through op that splits the mapped value into distinct outputs and then passes those outputs to your op that takes in two distinct arguments.

@op(out={"a": Out(Int), "b": Out(Int)})
def split_op(input_value):
    yield Output(output_name="a", value=input_value.get('a'))
    yield Output(output_name="b", value=input_value.get('b'))

@job
def test():
    my_op().map(lambda x: another_op(*split_op(x)))

For 2. You should be able to use a lambda in your mapping function:

@job
def test():
    mynumber = get_number()
    my_op_2().map(lambda x: another_op(a=x, b=mynumber))

The first sounds like a bit over-complication, I think I’ll just continue using a dictionary with my results instead which actually fits my use-case. The second point looks great, I’ll give it a try! Thanks a lot! 😀