How to pass external Asset to @ops in a graph via Op jobs?

[richban]

Context

I’m working on a pipeline where I need to load an external CSV file and process each row as a pd.Series through a series of operations in a graph. Each node in the graph may have external dependencies, such as reading geospatial data from files stored on S3. I want to represent these dependencies as external assets in Dagster, but without materializing them within Dagster itself (e.g., using xarray to read the geospatial data).

As each node processes its respective inputs, the external assets (like the geospatial data) will be passed between nodes as needed. Finally, the last node in the pipeline will handle the materialization of the final output

How would you approach structuring this in Dagster? Is this even possible?

Approach

Based on: https://www.youtube.com/watch?v=KVqyarPbCeU&t=795s

I want to use the burnable_4km asset as a reference to an S3 file path, without materializing it. Since burnable_4km is an external resource defined with AssetSpec and only want to reference the metadata (like the S3 file path) in my op.

# Define the burnable area as an external asset
burnable_4km_asset = AssetSpec(
    key=AssetKey(
        [EnvVar("BUCKET").get_value(), "wildfire", "burnable_4km"]
    ),
    metadata={
        "aws_account": EnvVar("AWS_ACCOUNT").get_value(),
        "s3_location": f"s3://{EnvVar('BUCKET').get_value()}"
        + "burnable_area.tif",
    },
    description="Wildfire location data of burnable 4km",
    group_name="external_assets",
)

@op
def read_burnable_area(
    context: OpExecutionContext,
    aws_session_resource: AWSSessionResource,
    burnable_4km # I believe this should be resolved by dagster
) -> xr.DataArray:
    s3_path = context.upstream_output.metadata["s3_location"]
    context.log.info(
        f"Reading burnable area from: {s3_path} for lat: , lon: "
    )
   # computation logic

# graph
@graph
def wildfire_damage_calculation():
    return read_burnable_area()


# definitions

defs = Definitions(
    assets=burnable_4km_asset,
    resources={
        "aws_session_resource": AWSSessionResource(
            aws_access_key_id=EnvVar("aws_access_key_id"),
            aws_secret_access_key="aws_secret_acess_key",
            region_name="aws_region",
        ),
    },
    jobs=[wildfire_damage_calculation.to_job()],
)

Issue

In Dagster UI I'm seeing the below error. What am I missing?

Missing required config entry "ops" at the root. Sample config for missing entry: {'ops': {'read_burnable_area': {'inputs': {'burnable_4km': '<selector>'}}}}

[richban]

I have found the answer here: #17718