在 Dagster 的 GraphQL API 中运行 ExecutePipeline 时 $repositoryLocationName 的值是多少？

Question

我正在尝试启动使用 GraphQL API 运行的 Dagster 管道。我在本地运行 Dagit，并有一个可以通过游乐场触发的工作管道。

但是，我现在尝试通过 GraphQL Playground 触发管道，可在

/graphql

获取。

我正在使用以下突变：

mutation ExecutePipeline(
  $repositoryLocationName: String!
  $repositoryName: String!
  $pipelineName: String!
  $runConfigData: RunConfigData!
  $mode: String!
)

...因此我提供以下查询参数：

{
  "repositoryName": "my_repo",
  "repositoryLocationName": <???>,
  "pipelineName": "my_pipeline",
  "mode": "dev",
  "runConfigData": {<MY_RUN_CONFIG>}
}

我不确定

repositoryLocationName

应该取什么值？我尝试了一些，但收到以下错误：

{
  "data": {
    "launchPipelineExecution": {
      "__typename": "PipelineNotFoundError"
    }
  }
}

这是我正在遵循的教程。

Answer 1

简短回答：

每个存储库都位于存储库位置内。如果您自己没有提供，Dagster 会提供默认存储库位置名称。要查找位置名称，您可以单击 Dagit 中的存储库选择器，它将位于存储库名称旁边：

在此示例中，存储库名称为

toys_repository

，位置名称为

dagster_test.toys.repo

更长的答案：

工作区（用您的

workspace.yaml

定义）是存储库位置的集合。

目前有三种类型的存储库位置：

• Python 文件
• Python模块
• gRPC 服务器

每个存储库位置可以有多个存储库。一旦您定义了位置，Dagster 就能够自动查找该位置中的所有存储库。在上面的示例中，我将工作区定义为具有单个 Python 模块存储库位置：

load_from:
  - python_module: dagster_test.toys.repo

请注意，只是指定了一个模块，并没有指定存储库位置名称，因此 Dagster 分配了一个默认的存储库位置名称。

如果我想指定一个位置名称，我会这样做：

load_from:
- python_module:
    module_name: dagster_test.toys.repo
    location_name: "my_custom_location_name"

对于 python 文件位置类似：

load_from:
- python_file: repo.py

或者使用自定义存储库位置名称：

load_from:
- python_file:
    relative_path: repo.py
    location_name: "my_custom_location_name"

Answer 2

您还可以使用 GraphQL 查询来查找。从文档中提供的示例开始，您只需添加

repositoryOrigin {
      repositoryLocationName
    }

导致

query PaginatedPipelineRuns {
  pipelineRunsOrError {
  __typename
     ... on PipelineRuns {
           results {
             runId
             pipelineName
             status
             runConfigYaml
             repositoryOrigin {
                repositoryLocationName
             }
          stats {
            ... on PipelineRunStatsSnapshot {
              startTime
              endTime
              stepsFailed
            }
          }
        }
      }
    }
  }

这将返回任何返回的运行的存储库位置名称。在查询之前触发您想要在 UI 中获取位置名称的管道，该运行将是您的第一个结果。

Answer 3

query RepositoriesQuery {
  repositoriesOrError {
    ... on RepositoryConnection {
      nodes {
        name
        location {
          name
        }
      }
    }
  }
}

无需任何参数。
这应该会产生如下响应：

{
    "data": {
        "repositoriesOrError": {
            "nodes": [
                {
                    "name": "__repository__",
                    "location": {
                        "name": "data-pipeline-1"
                    }
                }
            ]
        }
    }
}

从上面，我们可以形成repositoryName和locationName，如下所示：

{
  "repositoryName": "__repository__",
  "repositoryLocationName": "data-pipeline-1",
}

在上面的示例中，它是来自 dagster 云托管实例的响应。对于 dagster 云托管流程，存储库是内部变量

__repository__

。对于本地或自托管，我们应该观察响应并基于该 dagster 实例进行构建。