问题标签 [dagster]

作为输入，我想根据用户输入检索数据，或者如果没有给出用户输入，则从数据库中“随机”检索数据。管道的所有其他下游任务都是相同的。

因此，我想创建一个从实体 A 和 B 开始的管道，以及基于来自实体 A或实体 B的输入执行的下游实体 C。
但是，当在实体 A 和 B 上使用条件输出时，不执行实体 C，因为上游固体不产生一种输入。

有没有一种我错过的简单方法？

谢谢你的帮助。

我目前正在尝试在众多工作流框架之间进行选择。我需要一个重要的功能，即工作流组合。

即使在API 参考或高级教程中，我也没有在文档中找到任何内容。

所以我的问题是：是否可以组成管道？即用已经写好的有向无环图构建一些有向无环图。可能有一些解决方法，但我对它的原生集成很感兴趣。

谢谢

我正在构建一个原型管道，它做两件事：

1. （实心）从现有目录中清除文件
2. （实心）运行批处理以将数据转储到该目录中。

步骤#1 都是副作用，没有输出传递给#2。是否可以在管道中表达这两种实体之间的依赖关系？

我的项目结构如下所示：

• /definitions（适用于所有 dagster python 定义）
  • __init__.py
  • 存储库.py
  • /汇率
    • 管道.py
    • ...
  • ...
• 工作区.yaml

我尝试使用各种方法运行 grpc 服务器，尤其是以下方法（从项目根目录开始）：

• dagster api grpc -h 0.0.0.0 -p 4000 -f definitions/repositories.py
• dagster api grpc -h 0.0.0.0 -p 4000 -m definitions
• dagster api grpc -h 0.0.0.0 -p 4000 -m definitions.repositories

第一个命令产生以下错误：

dagster.core.errors.DagsterImportError：遇到 ImportError：attempted relative import with no known parent package从文件 C:\Users\Klaus\PycharmProjects\dagsterexchangerates\definitions\repositories.py 导入模块存储库时。考虑为基于-mCLI 的目标或python_packageworkspace.yaml 目标使用基于模块的选项。

第二个和第三个命令产生以下错误：

（堆栈跟踪在此之前）
ModuleNotFoundError：没有名为“定义”的模块

如何解决？

编辑：我已将我正在处理的示例的当前版本上传到 GitHub：https ://github.com/kstadler/dagster-exchangerates

EDIT2：反映目录结构的变化

我每天凌晨 12 点后 3 分钟触发@daily_schedule

由“2021-02-16 00:03:00”的预定刻度触发时

日期输入显示' 2021-02-15 00:00:00'，分区标记为'2021-02-15'

而如果通过分区“2021-02-16”的回填触发

日期输入显示' 2021-02-16 00:00:00'，分区标记为'2021-02-16'

为什么预定的刻度会在前一天填充分区？是否可以选择使用执行的日期时间（不使用 cron @schedule）？当我使用确切日期的时间戳执行查询时，这种差异令人困惑

PS我已经测试了预定运行和回填运行以具有相同的时区。

是否有一个字段可以从“上下文”中提取回溯堆栈以定位触发的异常并发送电子邮件或松弛消息？

在 Dagster 中创建 PartitionSetDefinition 时，您可以传入一个“模式”，该模式将交换所使用的资源（出于测试目的，您可能希望在 PROD 中使用云存储，但使用本地存储进行本地开发

模式要求您指定通常在环境 yaml 文件中提供的一组配置值，但是当您创建如下所示的 PartitionSetDefinition 时，您只能传递模式。这通常通过在管道上设置预设并将其用于运行来完成，但 PartitionSetDefinition 仅允许设置模式而不是预设。

您如何为此提供必要的预设/环境值？

我正在尝试使用以下方式安装 Dagster 和 Dagit：python3 -m pip install dagster dagit --user 但是，总是当它开始为 grpcio 构建轮子时，它会引发错误并停止安装。

有人在 Mac 上使用 Dagster 和 Dagit 吗？

编辑：我能够通过在我的 requirements.txt 文件中提供所需的版本来安装 Dagster。

我有一个由两个固体组成的 Dagster 管道（下面的可重现示例）。第一个 ( return_some_list) 输出一些对象的列表。第二个实体 ( print_num) 接受第一个列表（不是完整列表）中的一个元素，并对该元素进行一些处理。

我应该如何为第一个实体返回的列表的每个元素调用第二个实体？请解释任何最佳实践。

不确定这是否是最好的方法（让我知道），但我想print_num为第一个实体输出的每个元素生成一个不同的实体实例。这将帮助我在未来并行化实体并更好地处理长/计算密集型实体。

所以假设我有两个固体。第一个进行一些计算并将文件写入磁盘。第二个实体获取该文件并用它做其他事情，但它需要它的文件系统路径才能打开它。我可以使用两个yields （一个用于 the AssetMaterialization，另一个用于 the str Output）并明确地将 theOutput放入第二个可靠调用中：

这很好用，您应该在日志 ( 2021-03-16 13:23:29 - dagster - INFO - system - 366248ec-6a83-462f-b62f-9fb2514f6f80 - print_asset_path - /path/to/data/on/filesystem) 和AssetMaterializationin中获得信息消息dagit。

但是，这有点不方便，因为我需要使用我需要Output的文件系统路径显式生成一个。是否有可能以及如何AssetMaterialization在第二个实体中引用 并直接使用其属性？

类似的东西（不起作用）：