I'm trying to learn how to create a `composite_sol...
# announcements
k
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
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
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
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
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
great, thanks for filing! I’ll take a look tonight/tomorrow
k
Cool. Also ran into this related issue: https://github.com/dagster-io/dagster/issues/1608
m
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
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
.