Setting the file. One moment.
Subchapter 2.2
references/assets/advanced-patterns.mdMarkdown4 KBView on GitHub
Use when a single computation produces multiple assets. Define outputs with specs=[...] using AssetSpec, and yield MaterializeResult for each asset.
@dg.multi_asset(
specs=[dg.AssetSpec("users"), dg.AssetSpec("orders", deps=["users"])],
)
def load_data():
users_df = fetch_users()
yield dg.MaterializeResult(asset_key="users", metadata={"row_count": len(users_df)})
# orders depend on user data for enrichment
orders_df = fetch_orders(users_df)
yield dg.MaterializeResult(asset_key="orders", metadata={"row_count": len(orders_df)})When to use:
Subsettability: By default, all assets in a @multi_asset are materialized together. To allow materializing a subset, set can_subset=True on the decorator and skippable=True on individual AssetSpecs. Use context.op_execution_context.selected_asset_keys to check which assets were requested.
Static metadata on specs: AssetSpec accepts the same metadata parameters as @dg.asset — description, group_name, owners, tags, kinds, deps, code_version, automation_condition, etc. See Asset Definition Properties for details on each parameter.
MaterializeResult records dynamic metadata each time an asset materializes. Use it as a return type for @dg.asset or yield it in @multi_asset.
@dg.asset
def my_asset() -> dg.MaterializeResult:
data = [...]
return dg.MaterializeResult(
metadata={
"row_count": dg.MetadataValue.int(len(data)),
"last_updated": dg.MetadataValue.text(str(datetime.now())),
"sample_data": dg.MetadataValue.json(data[:5]),
}
)MaterializeResult[T] can also carry a value (like Output[T]), making it the preferred return type for greenfield code:
@dg.asset
def my_asset() -> dg.MaterializeResult[dict]:
data = {"key": "value"}
return dg.MaterializeResult(value=data, metadata={"size": len(data)})MetadataValue.int(n) — integer values (row counts)MetadataValue.float(n) — float values (percentages)MetadataValue.text(s) — short text valuesMetadataValue.json(obj) — JSON-serializable objectsMetadataValue.md(s) — markdown textMetadataValue.url(s) — clickable URLsMetadataValue.path(s) — file pathsMetadataValue.table(records) — tabular dataCompose multiple @ops into a single asset. Each op is independently retriable — if the last step fails, you can retry without re-running earlier steps.
@dg.op
def fetch_data() -> dict:
return {"raw": [1, 2, 3]}
@dg.op
def transform_data(data: dict) -> dict:
return {"processed": [x * 2 for x in data["raw"]]}
@dg.graph_asset
def complex_asset():
raw = fetch_data()
return transform_data(raw)When to use:
Combine @graph_asset and @multi_asset — compose ops into a pipeline that produces multiple assets.
@dg.graph_multi_asset(
outs={
"users": dg.AssetOut(),
"orders": dg.AssetOut(),
}
)
def etl_pipeline():
raw_data = extract_from_api()
cleaned = clean_data(raw_data)
return {"users": extract_users(cleaned), "orders": extract_orders(cleaned)}When to use:
deps=