问题标签 [prefect]

我正在尝试找到一种方法来更新开始用于调度流程的每次迭代的参数。

例如，假设我有一个计划在一年的每个星期一运行一次的流程。对于第一个星期一，流程需要以 5 的参数运行。下星期一需要以 7 的参数运行，以此类推。每周运行所需的参数将改变一个常数。

根据文档，看起来我可以为每次运行创建一个带有相应参数的时钟，但对于为许多运行安排的流来说，这似乎过多。

在 Prefect 中是否有更简单的方法？

我正在尝试从 teradata 获取数据，然后使用完美任务写入和读取镶木地板文件当我单独执行 sql 时，我的代码正在工作，如下所示

但是当相同的 fetcher 代码作为任务运行时，代码会失败

添加了 teradata 文件代码：

有人可以在这里建议/帮助吗？

如果本地代理正在运行，是否有一种机制可以自动注册流/新流，而不必手动运行，例如flow.register(...)在每个代理上运行？

在气流中，我相信他们有一个过程可以定期扫描dag指定气流主文件夹中名称中的任何文件，然后在它们中搜索 DAG 对象。如果找到它们，它会加载它们，这样就可以通过 UI 访问它们，而无需手动“注册”它们。

有没有类似的东西存在于省长身上。因此，例如，如果我刚刚创建了以下文件 test_flow.py，而不必运行它或添加它，flow.run_agent()是否有一种方法可以通过 UI 神奇地注册和访问它:) - 只是通过它简单地存在于适当的位置？

我可以编写一个与气流过程具有相似行为的脚本来定期扫描文件夹并注册流，但我想知道它是否有点hacky或者是否有更好的解决方案，我只是想太多了空气流动？

我已经建立了一个流程，如果 kwarg 为空，它会隐式跳过运行给定任务。

我在任务函数中使用这样的东西来跳过逻辑：

我想构建一些单元测试以确保跳过所述任务的最终状态。在任务级别获取状态的最简单方法是什么？

我可以从文档中看到如何获得流程而不是任务。

更新

为了补充克里斯的回应，我使用了他提出的第一个选项。由于我的流程是在测试之外定义的，因此我创建了一个简单的函数来获取一组已跳过的任务。在测试中，这与应该跳过的任务列表进行了比较：

我目前正在尝试在我的本地机器上运行一些完美的流程（然后在一些谷歌云运行实例中），我想将运行日志发送到远程的完美服务器。

这是示例代码：

另外，我在 backend.toml 中放置了一些配置：

所以，这是我遇到的问题：

当我运行这段代码时，我有这个错误=>

因为我的日志的第一行没有 flow_run_id。

示例： {"input": {"logs": [{"flow_run_id": null, "task_run_id": null, "timestamp": "2020-12-22T21:24:13.628595+00:00", "name": “prefect.FlowRunner”、“message”：“为“我的第一个流程”运行开始流程”、“级别”：“信息”....

我想知道我忘记了什么或者我做错了什么:(

感谢帮助

如果上游任务常量输入更改，我正在尝试构建一种机制来使任务的缓存无效。

我目前对每个任务的输入常量进行哈希处理，然后将其用作目标。我在想我可以建立当前任务输入以及上游任务输入的哈希值。正在考虑通过使用 Flow 对象中的常量和边来做到这一点。

这似乎是一个很常见的问题，想看看是否有一种简单的方法来实现它？

我已经开始与 prefect 合作，我正在尝试将我的结果保存到 Google 云存储：

如果我将 GOOGLE_APPLICATION_CREDENTIALS 环境变量设置为键，一切正常。

但是，在尝试对流程进行 docker 化时，我遇到了一些困难：

在这种情况下，当尝试使用 docker 代理运行我的流程时（无论是在注册流程的同一台机器上，我都会收到此错误）：

按照文档，我试图GCP_CREDENTIALS在我的 Prefect 云上设置一个秘密。但无济于事，我仍然遇到同样的错误。

我也尝试将结果保存在一个单独的GCSUpload任务中，但我仍然遇到同样的错误。

我看到的一种解决方案是通过 DockerFile 在我的 docker 映像中打包凭据，但是我觉得这应该是我应该使用 Prefect 机密的用例。

我以这种方式使用 Prefect 安排任务：

但是，如果文件在第一次 Prefect 运行后到达我的目录，则 file_names 在第二次运行期间保持为空，并且在所有下一次运行期间也是如此。

我试图用 grep 命令来获取我的文件，然后它就可以工作了！

有人知道为什么会这样吗？非常感谢您的帮助。

从Pycharm运行完美流程时，一切正常，但是当我从Prefect Server启动它时，流程找不到带有我的凭据的 .env 文件，并且由于此代码中的我自己的断言错误而失败：

我在我的虚拟环境（venv）上使用了这些命令来启动服务器和代理：

有任何想法吗？

我已经开始研究 python 中的完美工作流程，但没有找到任何有关批准工作流程的帮助。

谁能让我知道我将如何能够完美地创建审批工作流程？

谢谢