https://dagster.io/ logo
Powered by
k

King Chung Huang

07/25/2019, 7:58 PM
I'm trying to learn how to create a 
composite_solid
that uses another solid and sets a config value in it. In this code snippet, the 
prefix_value
solid prefixes the input value with the 
prefix
from the config. Then, the 
prefix_id
composite solid uses 
prefix_value
but sets the prefix to 
_id_
.
Copy code
@solid(
    config={
        'prefix': Field(str, default_value='_')
    }
)
def prefix_value(context, v):
    return f'{context.solid_config["prefix"]}{v}'


@composite_solid(
    config_fn=lambda _, cfg: {
        'prefix_value': {'config': {'prefix': '_id_'}},
    },
    config={},
)
def prefix_id(id):
    return prefix_value(id)
I'm getting 
dagster.check.CheckError: Invariant failed. Description: Cannot specify empty config for ConfigMapping
. But, I'm not sure how to solve it. Is this a correct way for a solid to use another solid with a custom config?
n

nate

07/25/2019, 8:44 PM
ah, that error message could be a lot better—the issue is that right now, the system doesn’t support an empty 
config
dictionary as you have it. so the quick solution is to just put some key in the config dictionary:
Copy code
@composite_solid(
    config_fn=lambda _, cfg: {
        'prefix_value': {'config': {'prefix': '_id_'}},
    },
    config={'a_key': Field(String)},
)
k

King Chung Huang

07/25/2019, 10:16 PM
Thanks. I put in a "foo" config so that it's not empty. But, I'm having trouble executing a pipeline with the composite solid.
Copy code
test_prefix cannot not be executed with the provided config. Please fix the following errors: Exception occurred during execution of user config mapping function <lambda> defined by solid prefix_id from definition prefix_id at path root:solids:prefix_id: Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/dagster/core/types/evaluator/evaluation.py", line 252, in _evaluate_composite_solid_config frozendict(context.config_value.get('config')), TypeError: 'NoneType' object is not iterable
I haven't been able to figure out what it's trying to iterate on that's None.
Here's my updated composite and the pipeline.
Copy code
@composite_solid(
    config_fn=lambda _, cfg: {
        'prefix_value': {'config': {'prefix': '_id_'}},
    },
    config={
        'foo': Field(str, is_optional=True, default_value='')
    }
)
def prefix_id(id):
    return prefix_value(id)

@pipeline
def test_prefix():
    # v = prefix_value()
    v = prefix_id()
    print_value(v)
n

nate

07/25/2019, 10:20 PM
hmm let me take a look and see what’s wrong
Copy code
# pylint: disable=no-value-for-parameter

from dagster import composite_solid, solid, pipeline, Field, execute_pipeline


@solid(config={'prefix': Field(str, is_optional=True, default_value='not defined')})
def prefix_value(context, some_id):
    return f'{context.solid_config["prefix"]}{some_id}'


@composite_solid(
    config_fn=lambda _, cfg: {'prefix_value': {'config': {'prefix': 'some override'}}},
    config={'foo': Field(str)},
)
def prefix_id(some_id):
    return prefix_value(some_id)


@pipeline
def test_prefix():
    prefix_id()


if __name__ == "__main__":
    result = execute_pipeline(
        test_prefix,
        {
            'loggers': {'console': {'config': {'log_level': 'DEBUG'}}},
            'solids': {
                'prefix_id': {
                    'config': {'foo': 'not used'},
                    'inputs': {'some_id': {'value': '12345'}},
                }
            },
        },
    )
this version works for me
k

King Chung Huang

07/25/2019, 10:36 PM
Ah, it works for me also when I include the config for prefix_id. Removing it results in the TypeError.
But, in my composite, I have the foo config set as optional. I was under the impression I wouldn't have to pass anything in for foo.
Here's what I have for code:
Copy code
from dagster import (
    Field,
    composite_solid,
    pipeline,
    solid,
)


@solid(
    config={
        'prefix': Field(str, is_optional=True, default_value='_')
    }
)
def prefix_value(context, v):
    return f'{context.solid_config["prefix"]}{v}'


@composite_solid(
    config_fn=lambda _, cfg: {
        'prefix_value': {'config': {'prefix': '_id_'}},
    },
    config={
        'foo': Field(str, is_optional=True, default_value='')
    }
)
def prefix_id(id):
    return prefix_value(id)


@solid
def print_value(context, v):
    <http://context.log.info|context.log.info>(str(v))


@pipeline
def test_prefix():
    # v = prefix_value()
    v = prefix_id()
    print_value(v)
Putting in an empty config works. I filed https://github.com/dagster-io/dagster/issues/1607 for this.
n

nate

07/25/2019, 10:59 PM
great, thanks for filing! I’ll take a look tonight/tomorrow
k

King Chung Huang

07/25/2019, 11:17 PM
Cool. Also ran into this related issue: https://github.com/dagster-io/dagster/issues/1608
m

max

07/25/2019, 11:53 PM
if i understand you correctly, and you're thinking about setting defaults for solid config at the pipeline level, not exactly. we do have two options which might give you what you're looking for -- the first is the notion of presets, which are config dicts that are set at the pipeline level (using the 
preset_defs
arg); the second is that you can execute a pipeline using composed environment dicts/yamls, which lets you put less-frequently changed config values in a single shared file
can you give us any color on the case you have in mind for this?
k

King Chung Huang

07/26/2019, 4:13 PM
I'm thinking about how I might translate an existing process into Dagster solids and pipelines. I have an action doing some text processing that is used multiple times in the same workflow with slighly varying configs. In a Dagster pipeline, using a solid multiple times with aliases makes sense. But, I wasn't sure how to set default configs for the solids. My first attempt was to use a composite solid, but the configs have been harder to work with than I expected for various reasons (issues filed). I'll take a look at 
preset_defs
.
Powered by