Start a backfill for all partitions from a sensor

[thomas-tran-de]

Hey, I want to start a backfill for all partitions of a partitioned run with a sensor.

I saw #19457 and tried the code below:

@dg.asset_sensor(
    asset_key=dg.AssetKey("trigger_asset"),
    job=target_job,
    default_status=dg.DefaultSensorStatus.RUNNING,
    minimum_interval_seconds=30,
)
def backfill_sensor(
    context: dg.SensorEvaluationContext,
) -> Generator[dg.RunRequest, None, None]:
    all_partitons = partition_def.get_partition_keys()
    yield dg.RunRequest(
        run_key=context.cursor,
        tags={
            "dagster/asset_partition_range_start": all_partitons[0],
            "dagster/asset_partition_range_end": all_partitons[-1],
            "dagster/backfill": context.cursor,
        },
    )

Unfortunately, I encountered several issues with this approach:

1. A single run starts instead of a backfill containing several runs.
2. The asset raises a DagsterTypeCheckDidNotPass stating that it got a dict as an input instead of the type that the asset expects. I guess this is a direct result of the single run instead of multiple runs for each partition being launched.

Alternatively, I know that it's possible to do

@dg.asset_sensor(
    asset_key=dg.AssetKey("trigger_asset"),
    job=target_job,
    default_status=dg.DefaultSensorStatus.RUNNING,
    minimum_interval_seconds=30,
)
def backfill_sensor(
    context: dg.SensorEvaluationContext,
) -> Generator[dg.RunRequest, None, None]:
    all_partitons = partition_def.get_partition_keys()
    for partition in all_partitions:
        yield dg.RunRequest(
            run_key=f"{context.cursor}-{partition}",
            tags={"dagster/partition": partition},
        )

This is not what I want since it creates many different runs that are not grouped under a single backfill.

Is there a good way to start a single backfill for all partitions from a sensor?

[ion-elgreco]

The first code runs in a single run backfill, which is how these partition start and end keys are used for.

You could use the graphql API from python to launch a backfill from the sensor.

Have you also looked at whether Automation conditions are sufficient enough instead of a sensor?

According to docs its now also possible to have custom automation condition

[thomas-tran-de]

Thanks for the answer. Using the GraphQL API sounds like the best shot. I specifically want to trigger a whole job, not specific assets. The way that I understood AutomationConditions is that they only work on an asset basis.

For future reference, the necessary GraphQL command is below:

mutation LaunchPartitionBackfill($backfillParams: LaunchBackfillParams!) {
  launchPartitionBackfill(backfillParams: $backfillParams) {
    ... on LaunchBackfillSuccess {
      backfillId
      __typename
    }
    ... on PartitionSetNotFoundError {
      message
      __typename
    }
    ... on PartitionKeysNotFoundError {
      message
      __typename
    }
    ... on InvalidStepError {
      invalidStepKey
      __typename
    }
    ... on InvalidOutputError {
      stepKey
      invalidOutputName
      __typename
    }
    ... on UnauthorizedError {
      message
      __typename
    }
    ... on PipelineNotFoundError {
      message
      __typename
    }
    ... on RunConflict {
      message
      __typename
    }
    ... on ConflictingExecutionParamsError {
      message
      __typename
    }
    ... on PresetNotFoundError {
      message
      __typename
    }
    ... on RunConfigValidationInvalid {
      pipelineName
      errors {
        message
        path
        reason
        __typename
      }
      __typename
    }
    ...PythonErrorFragment
    __typename
  }
}

fragment PythonErrorFragment on PythonError {
  message
  stack
  errorChain {
    ...PythonErrorChain
    __typename
  }
  __typename
}

fragment PythonErrorChain on ErrorChainLink {
  isExplicitLink
  error {
    message
    stack
    __typename
  }
  __typename
}

With the parameter looking something like this:

{
    "backfillParams": {
        "tags": [],
        "assetSelection": [
            {
                "path": [
                    "prefix",
                    "asset1"
                ]
            },
            {
                "path": [
                    "prefix",
                    "asset2"
                ]
            }
        ],
        "partitionNames": [
            "part1",
            "part2"
        ],
        "fromFailure": false,
        "selector": {
            "partitionSetName": "partitioned_job_name_partition_set",
            "repositorySelector": {
                "repositoryLocationName": "my_repo",
                "repositoryName": "__repository__"
            }
        }
    }
}

[thomas-tran-de]

I created a feature request (#27759) because I see multiple limitations with the current possibilities.