Asset dependencies across code locations

[zyd14]

What is the current canonical way to do this? Say I have a code location with an asset like this:

@asset
def A():
  ...

defs = Definitions(assets=[A])

and another code location with an asset B that is dependent on asset A

@asset
def B(A)
  ...
defs = Definitions(assets=[B])

The docs used to recommend using SourceAssets for this (before that I think it was asset sensors?), but references to SourceAssets seem to have been quietly removed from the docs. Is the new approach to use External Assets? Something else? Any ideas are very welcome!

[DennisSchwartz]

I'm also interested in this!

I have a small number of assets which require a different Python environment so I would like to separate them by code location.
The upstream asset A writes to S3 and the path is passed to downstream asset B via the Output and its metadata. The path itself is not static because it includes the data and code version of A which changes most runs.

When I try to specify A as the input to B I get an error:
Input asset "['pre', 'fix', 'A']" for asset "['pre', 'fix', 'B']" is not produced by any of the provided asset ops and is not one of the provided sources
Even when both code locations are loaded in the same workspace.

For now I have found a workaround by accessing the DagsterInstance and getting the latest materialisation event for asset A in the op of asset B. It's slightly complicated by the fact that my upstream asset is partitioned.

Here's what it looks like:

import pandas as pd
from typing import Any
from dagster import (
    AssetExecutionContext,
    AssetMaterialization,
    DagsterInstance,
    AssetKey,
    DagsterEventType,
    EventRecordsFilter,
    Output,
    SourceAsset,
    asset,
)

def get_metadata_per_partition(instance: DagsterInstance, asset_key: AssetKey) -> dict[str, Any]:
    ef = EventRecordsFilter(asset_key=asset_key, event_type=DagsterEventType.ASSET_MATERIALIZATION)
    mat_events = instance.get_event_records(event_records_filter=ef)
    all_materializations: dict[str, list[Any]] = {}
    for event in mat_events:
        mat = event.event_log_entry.dagster_event.event_specific_data.materialization
        partitions = mat.partition
        if partitions not in all_materializations:
            all_materializations[partitions] = []
        all_materializations[partitions].append(event)

    # Only keep the latest materialization for each partition
    latest_partitions: dict[str, dict] = {}
    for partition, materializations in all_materializations.items():
        latest: AssetMaterialization = max(materializations, key=lambda x: x.timestamp)
        metadata = latest.event_log_entry.dagster_event.event_specific_data.materialization.metadata
        latest_partitions[partition] = {
                k: v.value for k, v in metadata.items()
            }

    return latest_partitions



upstream_asset_key = AssetKey(["pre", "fix", "A"])

@asset(
    code_version="1",
    deps=[SourceAsset(key=upstream_asset_key)],
    name='B',
    key_prefix=["pre", "fix"],
)
def node_embeddings(context: AssetExecutionContext) -> Output:
    # Note: Dependencies across code locations seem to be not well-supported - this is a workaround to get the
    #       file paths of the upstream asset
    upstream_inputs = get_metadata_per_partition(
        context.instance, upstream_asset_key
    )
    nodes = pd.read_parquet(upstream_inputs["node"]["file_path"])
    # ... do something with the nodes
    

It's not exactly elegant, but it seems to work for me so far.

[alangenfeld]

references to SourceAssets seem to have been quietly removed from the docs

What made you say this? This section describes this case:

https://docs.dagster.io/concepts/assets/software-defined-assets#defining-external-asset-dependencies

[zyd14]

@alangenfeld What's the recommended approach to do this now that SourceAssets are being deprecated?

[psavalle]

@alangenfeld The link mentioned above seems to be dead now. https://docs.dagster.io/guides/build/assets/defining-assets-with-asset-dependencies#defining-asset-dependencies-across-code-locations mentions defining asset dependencies across code locations, but only with deps and when manually reading the input data, is there anything in the documentation that covers how to proceed if we want input assets from a different code location to be read via a standard IO manager?

[alangenfeld]

cc @cmpadden

[cmpadden]

Thanks @alangenfeld -

@psavalle it looks like the reference of using SourceAsset in that document was changed in this pull request: #21304.

If you only need to declare a dependency across code locations, the docs now recommend just depending on the upstream asset by asset key as referenced in https://docs.dagster.io/guides/build/assets/defining-assets-with-asset-dependencies.

If you need to model an upstream asset that Dagster doesn’t materialize, especially if you want Dagster to load it via an IO manager, the modern replacement for SourceAsset is AssetSpec.

If you have any recommendations on how to change the docs, please let me know.

[psavalle]

Thanks @cmpadden, looks like @dehume added a new section to the docs to cover that last week!

What remains missing in the docs is that as far as I can tell, you actually need to mirror all the metadata that may be needed by the IO manager, including partitions definition, metadata that specifies the database schema or partitioning database column, etc., right? The docs mention 'Cross-code-location dependencies on partitioned assets work the same way' but that's not really the case, and some guidance there would be useful.

This is unfortunately a pretty big limitation to using multiple code locations, as having to manually keep all the metadata in sync between the asset definition and the AssetSpecs in other code locations is tricky and can lead to subtle issues. It'd be ideal if Dagster was able to fill in this missing metadata directly (or, at least, flag when an asset definition in one code location is inconsistent with an AssetSpec in another code location but with the same key). This can be especially confusing as the lineage graph will consolidate the metadata (but then it might be missing when the IO manager is invoked when materializing the downstream asset).