Resource dependencies

[sryza]

This is surely not the first time this has been proposed, but the new AssetStore APIs make it a little more important.

Motivation

Example: Two AssetStores on same underlying storage system

A user has both of these in the same pipeline:

• An AssetStore that pickles objects and stores them in immutable per-run S3 files.
• An AssetStore that saves tables to known S3 locations as parquet files.

They’d like to be able to configure the S3 region in only one place. They’d also like to be able to mock S3 in one place.

Example: AssetStore + shared pipeline-level config

Each pipeline run processes a single partition of data. A user defines an AssetStore that needs to know the partition of the data it’s storing. The body of compute_fns also need to know the partition of the data they’re processing.

Proposal

@resource
def inner_resource(init_context):
    return MyResource(init_context.resources.other_resources)

@resource(required_resource_keys={'inner'})
def outer_resource(init_context):
    return MyResource(init_context.resources.inner)

@solid(required_resource_keys={'outer'})
def my_solid(_):
    pass

@pipeline(mode_defs=[ModeDefinition(resource_defs={
    'inner': inner_resource,
    'outer': outer_resource
})])
    my_solid()

Circular dependencies would result in an error at definition time.

Remapping

Not part of MVP, but want to make sure it’s possible in the future.

@resource
def inner_resource(init_context):
    return MyResource(init_context.resources.other_resources)

@resource(required_resource_keys={'inner'})
def outer_resource(init_context):
    return MyResource(init_context.resources.inner)

@solid(required_resource_keys={'outer_1'})
def my_solid_1(_):
    pass

@solid(required_resource_keys={'outer_2'})
def my_solid_2(_):
    pass

@pipeline(mode_defs=[ModeDefinition(resource_defs={
    'inner_1': inner_resource,
    'inner_2': inner_resource,
    'outer_1': outer_resource.remap_required_resource(old_key='inner', new_key='inner_2'),
    'outer_2': outer_resource.remap_required_resource(old_key='inner', new_key='inner_2'),
})])
    my_solid_1()
    my_solid_2()