问题标签 [prefect]

我有以下流程：

当我运行它时flow.run()，我收到以下错误：

是否可以迭代它，如果是，如何？

我有一个使用.map(); 因此，我“循环”了多个输入，但是一些输入我只需要生成一次，但我注意到我的流程不断重新生成它们。

是否可以在运行期间缓存/检查任务的结果（用于其他任务）？

我的理解是可以缓存特定的时间，如下所示：

但是，如果运行时间少于cache_for时间，缓存是否仍会保留下一次运行（如果不是，我猜长时间缓存会起作用）。

我的流程很长，我想将其状态标记为Failed特定任务失败。

我在这里阅读了文档，但是他们似乎没有指定这种情况。我的猜测是我需要在@task()装饰器中以某种方式具体说明这一点，但我不知道如何。

非常感谢任何指导。

我需要创建一个可以使用此工作流处理任务的批处理：

• 任务 #1 处理输入数据并返回列表列表。
• 任务 #2 从任务 #1 接收列表，并返回列表列表。
• 任务 #3 从任务 #2 接收列表，并返回列表列表。
• 任务#4 从任务#4 接收列表并处理列表中的数据。

例如，任务 #1 返回[[],[],[],[]]. 这意味着流程必须并行运行 4 个任务 #2。每个任务 #2 返回[[],[],[]]. 现在我们必须有 4x3 任务 #3。然后任务 #3 返回[[],[]]。最后流程必须运行 4x3x2 任务 #4。

可以使用 Prefect Flow 吗？我尝试使用映射功能，但它似乎不支持如此复杂的工作流模式（或者我可能没有正确使用它）。

当我运行流程 task1 返回正确数量的列表。然后 flow 创建了 4 个 task2，每个 task2 返回三个列表的列表。但是，流程并没有创建 12 个 task3，而是只创建了 4 个。每个 task3 接收 4 个列表的列表，因为它是使用 task1 而不是来自 task2 的 1 个列表创建的。

关于如何创建这样的工作流程的任何想法？

我的目标：

我有一个构建的 docker 映像，并希望在该映像上运行我的所有流。

目前：

我有以下任务在本地 Dask Executor 上运行。运行代理的服务器与执行所需的 Python 环境不同my_task- 因此需要在预构建映像中运行。

我的问题是： 如何在 Dask Executor 上运行此流程，以便它在我提供的 docker 映像（作为环境）上运行？

我认为我需要首先在该 docker 映像上启动服务器和代理（如此处所述），但我想有一种方法可以简单地在提供的映像上运行 Flow。

更新 1

按照本教程，我尝试了以下方法：

但这创建了一个图像，然后将其上传到registry_url提供的图像。之后，当我尝试运行注册的任务时，它会拉取新创建的图像，并且该任务现在停留在状态Submitted for execution几分钟。

我不明白为什么它会推送图像然后拉取它？相反，我已经在此注册表上构建了一个映像，我想指定一个应该用于执行任务的映像。

我在 Prefect 中有一个流程，其中 DB 客户端的启动和停止针对每个映射任务进行。下面是示例 ETL 结束的视觉效果。

save_transformed_data_to_db接受一个,unmapped(database)和unmapped(db_client)一个映射data值。

有没有办法将任务设置为在整个流程开始时只运行一次，在整个流程结束时只运行一次，如果是 - 如何？

Prefect 有这个Lazarus 进程，它“负责重新安排任何提交或运行的流程”。

这个问题是，如果您的重试延迟超过 10 分钟，您的流程将被 Lazarus 进程杀死。

有谁知道是否可以禁用或增加 Lazarus 进程的时间间隔以防止此类问题发生？

将 Prefect 配置为在本地运行时遇到困难。通过 pip3 安装，取决于 docker（已安装）和 docker-compose，已安装、已测试。作品。
我已经有一个正在运行的 Postgres 服务器，我喜欢配置为使用该服务器。但似乎在 docker 中也使用了一个？任何帮助表示赞赏。

运行时：

无法在本地找到图像 'docker/compose:1.27.4'
1.27.4：从 docker/compose 拉取（下载 en 开始运行 docker）

状态：为 docker/compose:1.27.4 下载更新的图像
警告：未设置 PREFECT_SERVER_DB_CMD 变量。默认为空字符串
警告：未设置 DB_CONNECTION_URL 变量。默认为空字符串
警告：未设置 POSTGRES_DB 变量。默认为空白
警告：未设置 POSTGRES_PASSWORD 变量。默认为空白字符串 >
警告：未设置 POSTGRES_USER 变量。默认为空字符串。

我正在尝试根据列表强制定义任务。挑战在于该列表应基于 Prefect 参数。

下面是我尝试过的代码，但显然它不起作用，因为 task_dependency_pairs 是一个任务，而不是一个列表。

如何在不破坏 Parameter 任务和其他动态生成的任务之间的依赖关系的情况下使其工作？

我正在为业务应用程序使用完美的工作流程。

我对存储在 postgresql 中的日志有疑问。

如果将日常日志保存在postgresql服务器上，数据量将是巨大的。

是否有一种机制可以轮换此日志并将其写入文本文件？

另外，将postgresql表中的数据写入文本文件等后，是否可以删除对应的记录？