问题标签 [dagster]

我最近选择了 Dagster 来评估作为 Airflow 的替代品。

我一直无法理解资源的概念，并试图了解我正在尝试做的事情是否可能或可以以不同的方式更好地实现。

我有一个像下面这样的帮助类，可以帮助保持代码干燥

s3_resource实际上是dagster_aws.s3.s3_resource，它将帮助我使用本地 aws 凭据连接到 AWS。

当我在下面的@resource部分进行调用时，我不确定如何将 s3_resource 传递给HelperAwsS3 。

请问有什么指点吗？还是我做错了，需要以不同的方式做？

谢谢你的帮助。

我正在使用 dagster 运行到本地 node.js 微服务管道，以便执行测试。

ide 是执行 n 个 docker_files 和 n 个 node.js 微服务，就像使用 dagster 一样轻松。

问题是，当我执行第一个第二个任务时，一个 shell 命令执行一个 docker 容器，dagsteer 保持在该点，而不是执行同一级别的所有任务。

像这样的当前 dag 日志

我可以同时执行所有 docker_elastic 和所有 node_services 吗？

是否有另一个易于配置的选项来轻松构建本地 dags？

谢谢

我刚开始关注dagster 教程。我设法hello_cereal使用 dagit 和 Python API 运行了这项工作，但由于某种原因，在尝试使用 dagster CLI 时

我得到一个DagsterUnmetExecutorRequirementsError：

确实，我没有$DAGSTER_HOME设置，但由于它适用于 Web UI 和 Python API 版本，我想知道我是否在其他地方犯了错误？

我在 macOS BigSur (11.6) 上，在全新的 miniconda 安装 ( v4.10.3)、Python 3.9.5 和 dagster 上0.13.0。

考虑这个例子 - 你需要从源数据库加载 table1，做一些通用的转换（比如转换时间戳列的时区）并将结果数据写入雪花。这是一个简单的，可以使用 3 dagster ops 来实现。

现在，假设您需要做同样的事情，但需要使用 100 张桌子。你会怎么用 dagster 做呢？你真的需要创建 100 个工作/图表吗？或者你可以创建一个作业，将执行 100 次？你能限制多少这些工作将同时运行吗？

我是 dagster 的新手，我正在尝试通过调用它的作业将资源传递给 dagster 操作，即使在遵循文档之后我也遇到问题，我不确定我是否需要再次将配置传递给作业似乎没有去工作。这是代码。

错误dagster.core.errors.DagsterInvalidConfigError: Error in config for job Error 1: Missing required config entry "resources" at the root.

我有一份 Dagster 工作正在训练 CNN（使用 Keras）。运行的 Opfit()导致以下错误：

没有给出额外的输出。我正在使用多容器本地 Docker 部署。

尝试的事情：

• 我通过使用在本地（非Docker）运行代码execute_in_process()，这可以正常工作。
• 由于在堆栈跟踪中提到了执行程序和多进程，我尝试将其设置为execution，in_process但这只是挂起。

任何建议将不胜感激。

我正在研究一个示例 dagster 管道。它应该从 Azurite Blobstorage 中挑选一些数据并将其写回到 Blobstorage 上的另一个文件夹中。

Blobstorage Azurite 也用于本地 MLflow 训练。

当我在本地执行 dagster 脚本时

它工作正常并且管道是绿色的，因此代码按预期运行。dagster 资源被加载，.csv 文件被加载并按预期保存。

但是，当我尝试在 dagster 的 dockerfile 中运行此代码（repositroy.py 文件和依赖项）时，它无法捕获数据。

Docker Compose 文件包含：

“general_nw”是 mlflow docker 容器的网络（见最后的链接）。

用于执行用户代码的 docker 是这样构建的：

当我从 docker_example_user_code 容器 ping azurite blobcontainer 时，我得到了响应。

所以我假设容器设置正确并且可以访问 azurite docker 容器。

我不想在此处添加整个代码，因为它在此azurite_dagster_pipeline链接下可用，并且您可以在此处找到带有其 azurite blobcontainer 的 local_mlflow 设置。

目前，这不是更大的 dagster 管道/教程的不工作基础。

我想我错过了在 docker-compose 或容器本身中设置变量。如果有人有一个想法或解决方案来修复 docker 环境中不工作的 dagster 管道，这将非常有帮助。

我希望你有所有的信息来提供一个解决方案。或建议。非常感谢您提前。

目前，我面临着dagster.core.errors.PartitionExecutionError来自 Dagster 的错误日志对我来说似乎并不明显。

我目前的设置是

我不确定我错在哪里，请您帮忙

在这里，我有 2 个问题。

1. 我需要运行一个 aiohttp 会话，该会话应同时向不同的 url 发出多个请求并下载多个文件并返回磁盘上这些文件的绝对路径列表。该列表应传递给另一个异步函数。

有没有办法在 dagster 作业中运行“async def”函数并构建异步管道？

2. 事实上，上述列表的长度可能因情况而异。每个文件都需要长时间而繁重的处理，并且由于处理阻塞（不幸的是），因此无法使其异步。因此，唯一的方法是在单独的线程或进程中启动此类处理，或者（就像我们所做的那样） - 在不同机器上的单独 docker 容器中启动此类处理。

dagster 是否可以使用操作动态创建 docker-containers，从它们返回任何输出并在容器退出时杀死它们中的每一个？

我有一个失败钩子，我想在其中发送一封邮件，但已引发异常。有什么方法可以访问它吗？