A software-defined asset is a named object in persistent storage—such as a table, file, or machine learning model—described in code together with how to produce and update it. Python functions declare the asset and its dependencies, allowing an orchestration system to schedule, execute, monitor, and trace data work. section 8.4
An asset definition connects several practical concerns:
- Storage target: load external CSV files into a DuckDB table or create data from another source. section 8.4.1
- Transformation: clean values, convert dates, add columns, or derive data with SQL and Python. section 8.4.3 section 8.4.4
- Dependency graph: declare prerequisites so execution is ordered and incomplete upstream assets produce warnings. section 8.4.3
- Operations: group assets into jobs, schedule them, materialize them manually, and inspect their state in the UI or CLI. section 8.4.1 section 8.4.2
Tennis data pipeline
Define an atp_matches_dataset asset that builds yearly CSV URLs and creates a typed matches table in DuckDB. Add atp_players_dataset to clean birth dates, then declare atp_players_name_dataset with a dependency that adds name_full. Dagster executes the upstream assets first and materializes the dependent table afterward. A command-line query can verify the result, such as the roughly 200,000 imported matches or the generated player names. section 8.4.1 section 8.4.3 section 8.4.2
Listing 8.15 Defining the first asset we want to process in our Dagster pipeline
from dagster_duckdb import DuckDBResource
from dagster import asset
@asset
def atp_matches_dataset(duckdb_resource: DuckDBResource) -> None:
base = "https://raw.githubusercontent.com/JeffSackmann/tennis_atp/master"
csv_files = [ #1
f"{base}/atp_matches_{year}.csv"
for year in range(1968,2024)
]
create_query = """
CREATE OR REPLACE TABLE matches AS
SELECT * REPLACE(
cast(strptime(tourney_date, '%Y%m%d') AS date) as tourney_date
) #2
FROM read_csv_auto($1, types={ #3
'winner_seed': 'VARCHAR',
'loser_seed': 'VARCHAR',
'tourney_date': 'STRING'
})
"""
with duckdb_resource.get_connection() as conn: #4
conn.execute(create_query, [csv_files]) #5
Inspect the complete Python definition of an asset that loads the ATP match files into DuckDB. section 8.4.1
Listing 8.18 A Dagster asset that transforms existing data
@asset(deps=[atp_players_dataset]) #1
def atp_players_name_dataset(duckdb: DuckDBResource) -> None:
concatenate_query = """
ALTER TABLE players ADD COLUMN name_full VARCHAR;
UPDATE players
SET name_full = name_first || ' ' || name_last #2
"""
with duckdb.get_connection() as conn:
conn.execute(concatenate_query, [])
Inspect the dependency declaration and SQL transformation that extend an existing players table. section 8.4.3
