hi 🙂 hopefully a simple question: how do i avoid re-evaluating assets when re-running jobs via the python API? for example, suppose the following example script, which defines an asset, defines a job using the asset, and runs the job with

execute_in_process()

:

# example.py

from dagster import asset, job
from dagster_gcp_pandas import BigQueryPandasIOManager
import pandas as pd

io_manager = BigQueryPandasIOManager(...)

@asset
def dummy_dataframe() -> pd.DataFrame:
    df = pd.DataFrame({"A": range(10)})
    df["current_time"] = pd.Timestamp.now()
    return df

@job(resource_defs={"io_manager": io_manager})
def my_job():
    dummy_dataframe()

if __name__ == "__main__":
    result = my_job.execute_in_process()

suppose i run this command twice:

$ DAGSTER_HOME=$(pwd) python example.py

my expectation is: if i run this script more than once, the job would use the persisted result for

dummy_dataframe()

from the first run. however, i can see in BigQuery that

dummy_dataframe()

is re-evaluated and updated each time i call the script (evidenced by the

current_time

field updating each run). why is this? why isn’t the job executor seeing that the asset was already computed and persisted, and simply loading that result? any help much appreciated 🙏

hi @William Grisaitis! a couple things to note here. The first is that in order to make a job that updates assets, you'll want to use define_asset_job, not the

@job

decorator (this is intended for use with ops/graphs, not assets). In your code, the dummy_dataframe asset is being invoked directly when the job is defined (rather than when

my_job

is being executed), and

my_job

actually does nothing, as it has zero ops inside of it.

Thank you so much @owen! I really appreciate your reply and setting me straight on the different job types.

No problem! To dive more into the programmatic invocation side, you'll need to start touching some non-public APIs. I don't anticipate any major changes to happen in this realm, but just keep in mind that things can shift around and get renamed without warning between minor releases. The way we calculate which assets are stale is with the CachingStaleStatusResolver. You can programmatically materialize a certain set of assets using

defs.get_implicit_job_def_for_assets(some_keys)

. putting this together, you could have a script that looked something like

defs = Definitions(assets=[all, your, assets])

instance = DagsterInstance.get()
asset_graph = defs.get_repository_def().asset_graph

# this bit relies on internal APIs
resolver = CachingStaleStatusResolver(instance, asset_graph)
stale_or_missing_keys: List[AssetKey] = []
for asset_key in asset_selection:
    if resolver.get_status(asset_key) in [StaleStatus.STALE, StaleStatus.MISSING]:
        stale_or_missing_keys.append(asset_key)

job = defs.get_implicit_job_def_for_assets(stale_or_missing_keys)

job.execute_in_process(instance=instance)