Using previous run's results

[szalai1]

I have an incrementally improving ML pipeline. Everytime the pipeline runs it should pick up the previous run's results and improve them.

The pipeline would be triggered by humans, so I cannot use dates to know the previous run.

Is there any way to get e.g the last successful run's run_id?

Thanks

[alangenfeld]

For the specific question of

Is there any way to get e.g the last successful run's run_id?

you can query the runs from the instance doing something like

instance = DagsterInstance.get()
runs = instance.get_runs(filters=PipelineRunsFilter(pipeline_name=pipeline_name, status=PipelineRunStatus.SUCCESS), limit=1)
# assert len(runs) == 1
last_success_id = runs[0].run_id

Fair warning though, at this stage of the project these instance APIs are still under development and may change.

Though I think the broader question points to a workflow that we could model better in the system cc @sryza

[szalai1]

Thank you, this will do it for now, if you have any suggestions on how could do it better, let me know.

[smackesey]

Update for current APIs 3 years later:

@op
def foo(context: OpExecutionContext):
    runs = context.instance.get_runs(
        filters=RunsFilter(job_name=context.job_name, statuses=[DagsterRunStatus.SUCCESS]),
    )

[brdunkley]

Another approach requesting only the previous run using run records.

import dagster as dg


@dg.op
def foo(context: dg.OpExecutionContext):
     run_filter = dg.RunsFilter(job_name=context.job_name, statuses=[dg.DagsterRunStatus.SUCCESS])

      # Retrieve the last run that matches the filter
      run_records = context.instance.get_run_records(
          filters=run_filter,
          limit=1,
          order_by="update_timestamp",  # Order by update time to get the most recent
          ascending=False  # Sort in descending order
      )

      # If there are no previous success return
      if len(run_records) == 0:
          return

      previous_run = run_records[0].dagster_run