Upgrading to software-defined assets#

This example is a vanilla, op-based job that follows the idiomatic practice of delegating all I/O to I/O managers and root input managers.

The goal of each op in the job is to produce an asset. However, because the job doesn't use the software-defined asset APIs, Dagster is unaware of this:

from pandas import DataFrame

from dagster import Definitions, In, Out, job, op

from .mylib import s3_io_manager, snowflake_io_manager, train_recommender_model


@op(
    ins={"raw_users": In(root_manager_key="warehouse")},
    out={"users": Out(io_manager_key="warehouse")},
)
def build_users(raw_users: DataFrame) -> DataFrame:
    users_df = raw_users.dropna()
    return users_df


@op(out={"users_recommender_model": Out(io_manager_key="object_store")})
def build_user_recommender_model(users: DataFrame):
    users_recommender_model = train_recommender_model(users)
    return users_recommender_model


@job(resource_defs={"warehouse": snowflake_io_manager, "object_store": s3_io_manager})
def users_recommender_job():
    build_user_recommender_model(build_users())


defs = Definitions(
    jobs=[users_recommender_job],
)

Here's what an equivalent job looks like using software-defined assets:

from pandas import DataFrame

from dagster import Definitions, SourceAsset, asset, define_asset_job

from .mylib import s3_io_manager, snowflake_io_manager, train_recommender_model

raw_users = SourceAsset(key="raw_users", io_manager_key="warehouse")


@asset(io_manager_key="warehouse")
def users(raw_users: DataFrame) -> DataFrame:
    users_df = raw_users.dropna()
    return users_df


@asset(io_manager_key="object_store")
def user_recommender_model(users: DataFrame):
    users_recommender_model = train_recommender_model(users)
    return users_recommender_model


users_recommender_job = define_asset_job(name="users_recommender_job")


defs = Definitions(
    assets=[raw_users, users, user_recommender_model],
    jobs=[users_recommender_job],
    resources={
        "warehouse": snowflake_io_manager,
        "object_store": s3_io_manager,
    },
)

This example does the same things as the previous example, with one difference. This job performs I/O inside of the ops instead of delegating it to I/O managers and root input managers:

from pandas import read_sql

from dagster import Definitions, In, Nothing, job, op

from .mylib import create_db_connection, pickle_to_s3, train_recommender_model


@op
def build_users():
    raw_users_df = read_sql("select * from raw_users", con=create_db_connection())
    users_df = raw_users_df.dropna()
    users_df.to_sql(name="users", con=create_db_connection())


@op(ins={"users": In(Nothing)})
def build_user_recommender_model():
    users_df = read_sql("select * from users", con=create_db_connection())
    users_recommender_model = train_recommender_model(users_df)
    pickle_to_s3(users_recommender_model, key="users_recommender_model")


@job
def users_recommender_job():
    build_user_recommender_model(build_users())


defs = Definitions(
    jobs=[users_recommender_job],
)

Here's an example of an equivalent job that uses software-defined assets:

from pandas import read_sql

from dagster import Definitions, asset, define_asset_job

from .mylib import create_db_connection, pickle_to_s3, train_recommender_model


@asset(non_argument_deps={"raw_users"})
def users():
    raw_users_df = read_sql("select * from raw_users", con=create_db_connection())
    users_df = raw_users_df.dropna()
    users_df.to_sql(name="users", con=create_db_connection())


@asset(non_argument_deps={"users"})
def user_recommender_model():
    users_df = read_sql("select * from users", con=create_db_connection())
    users_recommender_model = train_recommender_model(users_df)
    pickle_to_s3(users_recommender_model, key="users_recommender_model")


users_recommender_job = define_asset_job("users_recommenders_job", selection="*")


defs = Definitions(
    assets=[users, user_recommender_model],
    jobs=[users_recommender_job],
)

This example demonstrates a job where some of the ops (extract_products and get_categories) don't produce assets of their own. Instead, they produce transient data that downstream ops will use to produce assets:

from pandas import DataFrame

from dagster import Definitions, job, op

from .mylib import create_db_connection, fetch_products


@op
def extract_products() -> DataFrame:
    return fetch_products()


@op
def get_categories(products: DataFrame) -> DataFrame:
    return DataFrame({"category": products["category"].unique()})


@op
def write_products_table(products: DataFrame) -> None:
    products.to_sql(name="categories", con=create_db_connection())


@op
def write_categories_table(categories: DataFrame) -> None:
    categories.to_sql(name="categories", con=create_db_connection())


@job
def ingest_products_and_categories():
    products = extract_products()
    product_categories = get_categories(products)
    return write_products_table(products), write_categories_table(product_categories)


defs = Definitions(
    jobs=[ingest_products_and_categories],
)

Here's an equivalent job using software-defined assets.

Note: Because some ops don't correspond to assets, this job uses @op and @graph APIs and AssetsDefinition.from_graph to wrap a graph in a software-defined asset:

from pandas import DataFrame

from dagster import AssetsDefinition, Definitions, GraphOut, define_asset_job, graph, op

from .mylib import create_db_connection, fetch_products


@op
def extract_products() -> DataFrame:
    return fetch_products()


@op
def get_categories(products: DataFrame) -> DataFrame:
    return DataFrame({"category": products["category"].unique()})


@op
def write_products_table(products: DataFrame) -> None:
    products.to_sql(name="categories", con=create_db_connection())


@op
def write_categories_table(categories: DataFrame) -> None:
    categories.to_sql(name="categories", con=create_db_connection())


@graph(out={"products": GraphOut(), "categories": GraphOut()})
def ingest_graph():
    products = extract_products()
    product_categories = get_categories(products)
    return write_products_table(products), write_categories_table(product_categories)


two_tables = AssetsDefinition.from_graph(ingest_graph)


products_and_categories_job = define_asset_job(name="products_and_categories_job")


defs = Definitions(
    assets=[two_tables],
    jobs=[products_and_categories_job],
)

Ops and graphs#

Jobs#

Dagster types#

Code locations#

I/O managers#