将参数化的Prefect flow与随参数变化的明细表相结合
，可以实现更灵活和可扩展的数据处理和工作流程管理。
Prefect是一个开源的工作流程管理框架，它允许用户定义和管理复杂的数据处理流程。通过参数化Prefect flow，我们可以将流程中的一些参数抽象出来，使得流程可以根据不同的参数值执行不同的逻辑。这样，我们可以通过修改参数值来实现流程的定制化和灵活性。
而随参数变化的明细表是指根据不同的参数值生成的不同的明细表。在数据处理过程中，我们经常需要根据不同的参数值生成不同的数据集或者数据表，以满足不同的需求。通过将参数化的Prefect flow与随参数变化的明细表相结合，我们可以根据不同的参数值生成对应的明细表，并将其作为输入数据传入Prefect flow中进行处理。
这种结合的优势在于：
- 灵活性：通过参数化Prefect flow，我们可以根据不同的参数值执行不同的逻辑，从而实现流程的灵活定制和扩展。
- 可维护性：通过将参数化的Prefect flow与明细表相结合，我们可以将数据处理过程中的参数和数据分离，使得代码更易于维护和修改。
- 可扩展性：通过随参数变化的明细表，我们可以根据不同的参数值生成不同的数据集或者数据表，从而满足不同的需求，并且可以随时添加新的参数和明细表。
- 数据一致性：通过将参数化的Prefect flow与明细表相结合，我们可以确保不同参数值下生成的明细表与流程逻辑一致，避免数据处理过程中的错误和不一致。
应用场景：
- 数据处理：当需要对不同的数据集进行相同的处理逻辑时，可以通过参数化的Prefect flow与随参数变化的明细表相结合，实现批量处理不同的数据集。
- 工作流程管理：当需要根据不同的参数值执行不同的工作流程时，可以通过参数化的Prefect flow与随参数变化的明细表相结合，实现工作流程的定制化和灵活性。
推荐的腾讯云相关产品和产品介绍链接地址：
- 云函数（Serverless）：腾讯云云函数是一种事件驱动的无服务器计算服务，可以帮助用户在云端运行代码，无需关心服务器的管理和维护。通过云函数，可以方便地实现参数化的Prefect flow与随参数变化的明细表相结合的应用场景。详情请参考：腾讯云云函数
- 数据库（云数据库MySQL）：腾讯云云数据库MySQL是一种高性能、可扩展的关系型数据库服务，可以满足各种数据处理需求。通过云数据库MySQL，可以存储和管理随参数变化的明细表，并与参数化的Prefect flow相结合。详情请参考：腾讯云云数据库MySQL
- 对象存储（腾讯云对象存储COS）：腾讯云对象存储COS是一种安全、高可靠、低成本的云端存储服务，可以存储和管理各种类型的数据。通过对象存储COS，可以存储和管理参数化的Prefect flow所需的输入数据和输出结果。详情请参考：腾讯云对象存储COS
相关·内容
1回答
如果本地代理正在运行，是否有一种机制可以自动注册流/新流，而不需要手动运行(例如，在每个代理上运行flow.register(...) )？
在气流中，我相信它们有一个进程，它定期扫描指定气流主文件夹中的名称中的任何dag文件，然后搜索它们以查找DAG对象。如果找到它们，就会加载它们，这样就可以通过UI访问它们，而不必手动“注册”它们。
是否有类似的东西存在于总监。因此，例如，如果我刚刚创建了以下文件test_flow.py，而不一定要运行它或添加flow.run_agent()，那么是否有一种方法可以通过UI :)神奇地注册和访问它--仅仅通过它存在于适当的位置？
# prefect_hom
浏览 0提问于2020-11-23得票数 6
回答已采纳
TL;DR；
我无法使用prefect的来解决上述问题。我可能不是用错了(见下文)，就是漏掉了什么。真的很感激你的指点！
问题
我阅读了奇妙的，发现关于和的部分与此最相关(可能漏掉了一些东西！)在我看来，类是解决方案。
要查看是否可以使用Flow Runner恢复失败的流：
进行了失败的流运行：
from time import sleep
import prefect
from prefect import Flow, task
@task
def success():
sleep(3)
return
@task
def failure():
浏览 1提问于2020-07-27得票数 6
回答已采纳
1回答
在prefect中，我希望按计划运行带有参数的流。这听起来很自然地传递参数，例如db凭据，对吗？
我已经创建了一个具有与prefect run -p flow.py --param db_config_path=somepath一样运行良好的参数的流，现在我希望将它注册为prefect register --project project_name --path flow.py --name somename (服务器已经启动，代理正在运行)，以便按计划运行。
但是服务器上写着：graphql_1 | {"severity": "ERROR", "n
浏览 13提问于2022-04-14得票数 0
1回答
嗨，我试图应用到我的项目中，使用库click处理命令行的段落。下面是一个演示代码片段：
@click.command()
@click.option(
"-p",
"--pages",
type=int,
default=0,
help="...",
)
def main(pages):
print("Running...")
if pages > 0:
a()
else:
b()
print("Finished with
浏览 3提问于2021-12-13得票数 0
回答已采纳
在完美的工作流程中，我尝试持久化每次计划运行的数据。我需要比较每个先前和当前结果的数据。我尝试了Localresult和checkpoint=true，但它不起作用。例如,
from prefect import Flow, task
from prefect.engine.results import LocalResult
from prefect.schedules import IntervalSchedule
from datetime import timedelta, datetime
import os
import prefect
@task("func_task_
浏览 5提问于2021-03-17得票数 1
1回答
我正在使用并定义一个来使用cosmos db插入文档。
问题是调用是可迭代的，对于大型容器，无法将所有条目保存在内存中。
我相信我的问题可以归结为：
given an iterator, how can I create batches to be processed (mapped) in a prefect flow?
示例：
def big_iterable_function_i_cannot_change():
yield from range(1000000) # some large amount of work
@task
def some_prefect_
浏览 1提问于2021-03-16得票数 0
1回答
我正在尝试使用实现一个完美的流
from prefect import Flow, Parameter
from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock
a = Parameter('a', default=None, required=False)
b = Parameter('b', default=None, required=False)
schedule = Schedule(clocks=[
CronClock(
浏览 2提问于2020-06-24得票数 8
回答已采纳
我希望能够重用相同的参数化流，但因输入而异。举个例子：
from prefect import task, Flow, Task, Parameter
from prefect.schedules import CronSchedule
diurnal = ['rooster', 'dog']
nocturnal = ['owl', 'hampster']
# Schedules
diurnal_schedule = CronSchedule("50 7 * * mon,wed")
nocturnal_
浏览 8提问于2019-12-04得票数 3
我正在尝试用一个prefect流包装一个类，该流目前通过类似于以下内容的类工作：
class Algorightm:
@task
def do_first(self, something):
....
@task
def do_second(self, something_else):
....
@flow
def start_flow(self):
do_first(..)
do_second(..)
通常情况下，这是正常的，如果我将其分解为外部函数，流就会正常运行，这意味着如果我只在类的外部使用它
浏览 16提问于2022-10-23得票数 1
1回答
我已经开始使用prefect，我正在尝试将我的结果保存到Google云存储： import prefect
from prefect.engine.results import GCSResult
from prefect.run_configs import DockerRun, LocalRun
from prefect.storage import Docker, Local
@prefect.task(checkpoint=True, result=GCSResult(bucket="redacted"))
def task1():
return 1
s
浏览 21提问于2021-01-19得票数 0
我修改了简单的示例‘收藏编号. the’：
from prefect import flow
@flow
def my_favorite_function(myvar:str='1'):
print(f'\n\nLOCALS: {locals()}\n\n')
在代理端，输出是：
Agent started! Looking for work from queue 'Agent queue semua'...
12:51:44.174 | INFO | prefect.agent - Submitting flow run
浏览 7提问于2022-08-29得票数 1
回答已采纳
1回答
我是新来的省长，并希望部署在Azure Kubernetes服务使用省长库伯内特斯的工作。我已经建立了一个Linux容器，包括Prefect，kubectl。我可以建立到AKS集群的连接，通过使用kubectl，我可以在AKS部署一个Kubernetes作业。但是它是如何与省长合作的呢？在容器中，我存储了.py文件和YAML文件，后者定义了Kubernetes作业。
-> kubectl应用-f deploytestcontainer.yaml -> works
运行附加示例Prefect代码也可以(但作业不会在AKS上部署)
这就是"firstk8sj
浏览 8提问于2022-02-19得票数 0
回答已采纳
1回答
如何在码头图像上执行完美流？
、、、、
我的目标：
我有一个构建的码头映像，并希望运行我的所有流在该映像。
当前：
我有以下任务，它运行在本地Dask执行器上。代理运行的服务器与执行my_task所需的python环境不同，因此需要在预构建映像中运行。
，我的问题是：如何在Dask执行器上运行这个流，以便它在我提供的对接映像上运行(作为环境)？
import prefect
from prefect import task, Flow
from prefect.engine.executors import LocalDaskExecutor
from prefect.environments import LocalEnviron
浏览 0提问于2020-10-07得票数 7
回答已采纳
1回答
完善进度延迟运行
、、、
我有一个简单的‘你好世界！’有时间表的代码。但是它们总是很晚才开始，黄色的矩形被放置在图中，如屏幕截图()所示。
有什么问题吗？为什么会出现“迟跑”？А以及如何使绿色长方形代替黄色？
from prefect import task, Flow
from datetime import timedelta
from prefect.schedules import IntervalSchedule
@task
def say_hello():
print("Hello, world!")
schedule = IntervalSchedule(interval=
浏览 10提问于2022-04-19得票数 0
摘要
我在试着理解码头长官的特工。为此，我试图在本地计算机上配置一个最小的设置。我已经设法让docker代理连接到本地服务器，看起来它正在运行流。但是，似乎在流完成后，代理无法在服务器上更新流的状态，因为它无法连接到服务器后端。
详细信息
这是我的流程：
import prefect
from prefect import task, Flow
from prefect.run_configs import DockerRun
@task
def say_hello():
logger = prefect.context.get("logger")
logge
浏览 5提问于2021-11-16得票数 1
回答已采纳
目前，我正试图在本地机器上运行一些优秀的流(然后在一些Google云运行实例中)，并且我希望将运行日志发送到远程的prefect服务器。
下面是示例代码：
@task
def my_task():
logger = prefect.context.get("logger")
logger.info("An info message.")
logger.warning("A warning message.")
@task(log_stdout=True)
def say_hello(name):
print(&
浏览 3提问于2020-12-22得票数 4
这看起来应该很简单，但我不知道如何在单个流中包含状态和数据依赖关系。以下是我尝试(简化)的内容：
def main():
with Flow("load_data") as flow:
test_results = prepare_file1()
load_file1(test_results)
participants = prepare_file2()
load_file2(participants)
email = flow.add_task(EmailTask(name='ema
浏览 5提问于2022-03-05得票数 0
我一直在使用Prefect进行工作流管理，但在使用Prefect的资源管理器建立和停止spark会话时遇到了麻烦。 我浏览了Prefects文档，下面提供了一个使用Dusk的示例： from prefect import resource_manager
from dask.distributed import Client
@resource_manager
class DaskCluster:
def init(self, n_workers):
self.n_workers = n_workers
def setup(self):
&
浏览 23提问于2021-08-20得票数 2
1回答
我目前正在使用Python Prefect (prefect.io)，我编写了下面的代码
from prefect import Flow, task
@task
def say_hello():
print('Hello')
@task
def say_how_a_u():
print('How are you?')
@task
def say_bye():
print('Bye Bye')
with Flow('Test') as flow:
say_hello()
say_how_a
浏览 10提问于2022-03-24得票数 0
回答已采纳
我正在尝试创建一个由文件处理程序创建的Prefect.io日志文件，使其看起来与Prefect.io记录器打印出的日志完全相同，但我得到的只是消息。
而不是
[2021-06-11 12:46:02-0700] INFO - prefect.FlowRunner | Beginning Flow run for 'test'
[2021-06-11 12:46:02-0700] INFO - prefect.TaskRunner | Task 'pull_project': Starting task run...
[2021-06-11 12:46:02-07
浏览 12提问于2021-06-12得票数 2
大家好，我试图使用prefect为python创建一个流，但是我得到了错误TypeError：'fn‘必须是可调用的，我已经安装了所有的包，并且我尝试过在不同的IDE中安装，比如vs代码和colab，但是结果是一样的。
守则是：
from prefect import task, Flow
@task()
def load():
print("Loading data")
with Flow("Hello-world") as flow:
load()
flow.run()
#ERROR
TypeError: 'fn
浏览 25提问于2022-07-31得票数 4
import time
from prefect import task, flow
@task
def my_task():
return 1
@flow
def my_flow():
my_task.submit()
if __name__ == "__main__":
my_flow()
我正在尝试运行这个简单的流，但是我总是在下面看到这个错误。我用的是prefect==2.2.0
浏览 6提问于2022-08-24得票数 0
1回答
在阅读了关于的文档之后，我认为这个工作流应该是输出缓存的一个例子：
from time import sleep
from prefect import Flow, task
from prefect.engine.results import LocalResult
@task(target="func_task_target.txt", checkpoint=True,
result=LocalResult(dir="~/.prefect"))
def func_task():
sleep(5)
return 99
wit
浏览 0提问于2021-06-09得票数 2
回答已采纳
我正在尝试创建一个Prefect任务，它接收PyMySQL connection的一个实例作为输入，例如：
@task
def connect_db():
connection = pymysql.connect(user=user,
password=password,
host=host,
port=port,
db=db
浏览 14提问于2021-06-12得票数 1
使用Prefect向我的后端服务器注册工作流将返回以下错误：
追溯(最近一次调用)：文件"/home/usr/Documents/Test_Automation/Automation/qa-automation/date_validator_Prefect/date_validator.py"，行114，在flow.register(project_name="QA-Test")文件"/home/usr/.local/lib/python3.9/site-packages/prefect/core/flow.py"，第1726行中，在寄存器re
浏览 5提问于2022-05-30得票数 0
回答已采纳
如何使用不同的参数多次运行流？我试着像这样使用create_flow_run.map：
dump_to_gcs_flow = create_flow_run.map(
flow_name=unmapped(utils_constants.FLOW_DUMP_TO_GCS_NAME.value),
project_name=unmapped(constants.PREFECT_DEFAULT_PROJECT.value),
parameters=tables_to_zip,
labels=unmapped(current_flow_labels),
run
浏览 3提问于2022-09-09得票数 1
回答已采纳
1回答
我的程序总是在启动时运行，而不是预定的时间，也许我只是误解了什么，但我不能说为什么它会这样做。
from prefect import flow, task, get_run_logger
from prefect.deployments import Deployment
from prefect.orion.schemas.schedules import (
CronSchedule,
IntervalSchedule,
RRuleSchedule,
)
@task(retries=2, retry_delay_seconds=5)
def say_hello():
浏览 11提问于2022-09-02得票数 -1
1回答
使用，我想从另外两个流创建一个新流。
我遇到的错误是A task with the slug "add_num" already exists in this flow.，是否有可能更新使用相同tasks或Parameters的Flows。下面是我想要完成的最简单的例子。`
from prefect import task, Flow, Parameter
@task
def add_one(x):
return x+1
with Flow("Flow 1") as flow_1:
add_num = Parameter("
浏览 0提问于2020-03-14得票数 5
回答已采纳
我正在探索Prefect的map-还原功能，它是一个强大的成语，用于编写大规模并行、健壮的外部数据导入程序。
作为一个例子--非常类似于 --请考虑下面的片段：
@task
def retrieve_episode_ids():
api_connection = APIConnection(prefect.context.my_config)
return api_connection.get_episode_ids()
@task(max_retries=2, retry_delay=datetime.timedelta(seconds=3))
def download_
浏览 2提问于2022-04-28得票数 0
交叉张贴这个问题：
我有一个使用Prefect v1.3.1的最小项目，您可以在这里找到它：。在从本机代码迁移到坞化代码之后，我在获得正确的部署方面遇到了一些困难。我正在使用Prefect v1的最新版本，在我能够迁移到Prefect 2.x之前需要解决这个问题
我正在使用诗歌，但我很确定这里有一个模式，适用于Hatch，Pipenv，Flit或其他工具。在注册过程中观察到了这个问题，它将记录到最后：
ModuleNotFoundError: No module named 'aid'
The command '/bin/sh -c python /opt/prefe
浏览 27提问于2022-09-17得票数 1
回答已采纳
我有一个完美的服务器在本地运行(0.13核心版本)。我在一个服务器机器上循环调用了flow.run() 1000次，这个服务器机器有64 GB内存和32个核心。当它运行到大约300次时，它开始从GraphQL抛出连接被拒绝的错误。 我还在考虑是否在我的工作流程中使用Prefect，但它看起来占用了太多的RAM。Prefect如何同时扩展数千个工作流？ 我正在用一个简单的例子运行工作流： 176 from flask import Flask
177 app = Flask(__name__)
178
179 import prefect
180 client = prefect.Client
浏览 22提问于2020-08-07得票数 7
我如何才能添加一个记录器到prefect的prefect.tasks.database.sqlite.SQLiteQuery类中，就像使用@task装饰器一样：
from prefect import task, Flow
import prefect
from time import sleep
@task()
def some_task():
logger = prefect.context.get("logger")
logger.info("Let's sleep a second!")
sleep(1)
vers
浏览 8提问于2022-07-07得票数 0
回答已采纳
1回答
当打开"python.analysis.typeCheckingMode": "strict"或"python.analysis.typeCheckingMode": "basic"时，包含其他模块导入的.py文件不识别重新导入，并以错误突出显示它们：
"Flow" is not exported from module "prefect"
Import from "prefect.core" insteadPylancereportPrivateImportUsage
如果原始
浏览 2提问于2022-01-17得票数 4
我已经建立了一个systemd服务来运行基于这个文章的Linux代理，但是它找不到dbt命令。有人知道如何让服务捕获dbt命令吗？
我正在使用GCP，Ubuntu，20.04LTS，amd64
以下是服务状态：
prefect-agent.service - Prefect Agent
Loaded: loaded (/etc/systemd/system/prefect-agent.service; disabled; vendor preset: enabled)
Active: active (running) since Tue 2023-03-28 02:03:0
浏览 0提问于2023-03-28得票数 0
回答已采纳
当使用docker作为存储时，在调用中如下：
prefect register --project yourproject -p yourflow.py
似乎perfect创建了一个tmp{randomNumber}文件夹，并放置了参与docker构建的healthcheck.py和docker文件。
但是我不小心删除了tmp文件夹，即使我恢复了它，它仍然抛出以下错误。发生的事情是Prefect期望一个以新的random_number作为后缀的tmp文件夹。我如何才能从中恢复呢？
Step 10/11 : COPY ./tmpnu2k398p/healthcheck.py /opt/prefe
浏览 3提问于2021-12-02得票数 0
1回答
我用这种方式和省长安排任务：
#Python script
from prefect import task, Flow
from prefect.tasks.shell import ShellTask
from datetime import timedelta
from datetime import datetime
from prefect.schedules import IntervalSchedule
import os
import sys
schedule = IntervalSchedule(start_date=datetime.now() + timedelta(s
浏览 6提问于2021-01-26得票数 0
我刚刚开始阅读关于省长(并有一点使用气流的经验)。
我的目标是设置一项任务，每天在Prefect中运行，并将数据收集到一个文件夹(我猜这就是Prefect可以帮助我完成的任务)。另外，我的目标是填充历史数据(就像我把这个作业运行回原来的时间一样)。
在气流中，有一个start_date的概念，当在过去设置时，DAG将从那个日期开始运行，并在每个间隔上填充。
例如，如果我有一个任务获取日期并返回该日期的数据，如：
# Pseudo code
def get_values_from_somewhere(date: datetime) -> dict:
return fetched_v
浏览 0提问于2020-09-23得票数 7
回答已采纳
1回答
如何模拟在同一个文件中使用不同参数两次调用的函数？下面是我的代码片段：
code_file.py
from prefect import Client
client = Client(
api_server = <ip of server>
)
def run_flow(name1, name2):
# some graphql queries, not shown here
value1 = client.graphql(graphql_value1, variables={'name': name1})
print("
浏览 12提问于2022-03-24得票数 0
回答已采纳
我有一个任务可以确保一些日志文件是空的(下面是task4)，但是在我的当前流中，它首先作为根任务运行，即使我在最后尝试运行它的逻辑时：
with Flow('Pipeline') as flow:
bool1 = task1.run()
bool2 = task2.run()
bool3 = task3.run()
if bool1 and bool2 and bool3:
task4.run()
所以我觉得州长不关心条件逻辑。是否有办法使任务1,2,3上游任务为task4？我唯一能想到的方法是将bool变量作为参数引入task
浏览 9提问于2022-07-26得票数 1
1回答
我目前在一个简单的python类中定义了一个工作流。在这个类中，我为每个任务定义了一个单独的方法。我现在通过class.run()启动ETL。
class Main():
...
def _read_data(self, fname: str):
""" read data from file """
....
return data
def _process_data(self, data):
""" process data ""
浏览 7提问于2022-03-14得票数 0
我的工作流中有两个任务A和B，如果任务失败，我想跳过B，请注意吗？@任务A()：Pass B()：Pass (“”)作为flow：
浏览 7提问于2022-04-10得票数 0
回答已采纳
1回答
pub struct Dest<'a> {
pub data: Option<&'a i32>,
}
pub struct Src<'a> {
pub data: Option<&'a i32>,
}
pub trait Flowable<'a: 'b, 'b> {
fn flow(&'a self, dest: &mut Dest<'b>);
}
impl<'a: '
浏览 0提问于2018-12-14得票数 2
回答已采纳
1回答
我有一个特定的ETL问题:在我的例子中，数据来自分页资源，并且可以被固定大小的页面获取。因此，对API的调用指定了第1页、第2页、第3页等。根据我的计划，每个页面都会被转换和加载。因此，页面的ETL是(原子(页)任务的etl_batch流)，但是我想在整个分页资源上运行ETL，直到最后一个可用的页面。因此，我认为应该在etl_batch流上(每隔几秒钟调用一次)通过一个循环(例如每小时调用一次)来解决这个问题。基于下面的参考文献，我认识到模型必须被转换成(页面)任务流的一个流程。
下面是我遇到的问题:如果我指定了etl_batch流的计划，那么它的状态就会被重置。但是我注释掉了etl_ba
浏览 3提问于2022-04-12得票数 1
1回答
我试图使用存储在prefect cloud中的prefect cloud和coiled的秘密来运行coiled。
用一个flow输出prefect client error错误
prefect.utilities.exceptions.ClientError: Malformed response received from Cloud - please ensure that you have an API token properly configured.
这是身份验证错误吗？我是否需要将一个prefect cloud authentication token传递给coiled
浏览 12提问于2021-03-16得票数 0
回答已采纳
我计划从Revit文件中读取明细表数据，如(房间明细表或门明细表)。有没有使用revit的设计自动化api来做这件事的例子？
浏览 4提问于2020-03-27得票数 0
我能够通过cli成功地创建一个函数应用程序，但是当我去发布/创建函数代码时，我会得到标题中的错误。
我将按照这个文档来创建带prefect的blob触发器函数，所以这就是为什么我的代码中还有另外两个配置设置。
az functionapp config appsettings set `
--name PrefectBlobTrigger `
--resource-group my_resource_group`
--settings "BlobConnectionString=my_blob_connection_str PREFECT_API_URL=my_prefect_api_
浏览 9提问于2022-12-01得票数 1
回答已采纳
1回答
我想实施一个中央主管项目，随着时间的推移，可以相互独立地增加流量。该项目的结构如下：
prefect/
├── src/
│ ├── flows/
│ │ ├── test_pack1/
│ │ │ ├── common/
│ │ │ │ ├── __init__.py
│ │ │ │ └── test_module.py
│ │ │ ├── .env
│ │ │ ├── __init__.py
│ │ │ ├── requirements.txt
│ │ │ └── test_pack1_
浏览 4提问于2022-04-08得票数 2
我已经为sys.excepthook设置了一个异常处理程序
def handle_exception(exc_type, exc_value, exc_traceback):
if issubclass(exc_type, KeyboardInterrupt):
logger.error("KeyboardInterrupt\n")
sys.__excepthook__(exc_type, exc_value, exc_traceback)
return
# Pre
浏览 7提问于2021-12-21得票数 1
对于我来说，从自定义的科学工作流管理(python)转向一些团队工作的时间已经过去了。简而言之，我的工作流程涉及长时间运行的(天)流程，其中包含大量共享参数。作为依赖图，节点是产生输出或执行某些其他工作的任务。这在工作流工具中似乎相当普遍。 然而，我需要的关键是每个任务都是由它所需的参数定义的。根据这些参数及其依赖项的所有参数的状态实例化任务。因此，如果一个任务已经根据给定的参数状态完成了它的作业，那么它就是完成的，并且不会重新运行。此参数状态不是全局参数状态，而只是与DAG的该部分相关的状态。这种对参数状态而不是完成时间的依赖似乎是我的需求和现有工具之间的本质区别(至少我从Luigi和Ai
浏览 22提问于2020-03-31得票数 2
1回答
下面是我死去的简单流程：
from prefect import flow
import datetime
@flow
def firstflow(inreq):
retval={}
retval['type']=str(type(retval))
retval['datetime']=str(datetime.datetime.now())
print(retval)
return retval
我负责猎户座长官和警长。
使用web (部署运行)创建一个触发器.代理人成功地拉出并完成了这项工作。
我的问题是怎样用卷发来做触发？注意:
浏览 5提问于2022-08-16得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
