Ops, Jobs and Resources

Not all work produces an asset. An op is a plain unit of computation, and a job wires ops into a graph by passing outputs to inputs, much like TaskFlow (The TaskFlow API and XComs). Ops were Dagster 177,056 's original model; assets are built on them. Both receive resources: typed, configurable objects such as database clients, injected by parameter name and swapped in tests, so no function reads a password:

definitions.py (excerpt): a resource, two ops and a jobPython

class BookNestDB(dg.ConfigurableResource):
    host: str = "l2-pg"
    port: int = 5432
    password: str
    def connect(self):
        return psycopg2.connect(host=self.host, port=self.port, dbname="booknest",
                                user="postgres", password=self.password,
                                options="-c TimeZone=UTC")
@dg.op
def table_sizes(db: BookNestDB) -> dict:
    ...                                          # {table: bytes} for orders and order_items
@dg.op
def analyze_tables(context: dg.OpExecutionContext, db: BookNestDB, sizes: dict) -> None:
    conn = db.connect()
    conn.autocommit = True                       # VACUUM cannot run inside a transaction
    with conn.cursor() as cur:
        for table, size in sizes.items():
            cur.execute(f"VACUUM (ANALYZE) {table}")
            context.log.info(f"vacuumed {table} ({size / 2**20:.1f} MiB)")
@dg.job
def booknest_maintenance():
    analyze_tables(table_sizes())                # data flow between ops is the dependency

The Definitions object at the bottom of the module binds db to BookNestDB(password=dg.EnvVar("BOOKNEST_PG_PASSWORD")), read when a run starts. dagster job execute -j booknest_maintenance ran both ops and logged vacuumed orders (11.9 MiB) and vacuumed order_items (10.3 MiB).

Use ops for side effects; model anything a consumer reads as an asset, or Dagster cannot show its lineage.

MENU