Python 3.0 时代的工作流编排王者:Prefect 3.0

from prefect import flow, taskimport httpx@task(log_prints=True)def fetch_stars(repo: str):    url = f"https://api.github.com/repos/{repo}"    data = httpx.get(url).json()    stars = data["stargazers_count"]    print(f"{repo} has {stars} stars!")    return stars@flow(name="GitHub Stars Tracker")def github_flow(repos: list[str]):    for repo in repos:        fetch_stars(repo)if __name__ == "__main__":    github_flow(["PrefectHQ/prefect", "pandas-dev/pandas"])

# 通过 Prefect SDK 创建自动化规则automation = await prefect_client.create_automation(    name="pause-on-failure",    trigger=Trigger(        event="prefect.flow-run.failed",        match={"deployment.name": "critical-workflow"}    ),    action=PauseDeploymentAction())

from prefect.events import emit_eventemit_event("data.pipeline.completed", payload={    "records_processed": 10000,    "status": "success"})

from prefect import flow, taskfrom prefect.transactions import transaction@taskdef deduct_inventory(item: str, qty: int):    print(f"Deducting {qty} of {item}")    return {"deducted": True}@task(on_rollback=lambda: print("Rolling back inventory..."))def write_to_warehouse(data):    print(f"Writing {data} to warehouse")    return {"committed": True}@flowdef order_flow(item: str, qty: int):    with transaction():        inventory_result = deduct_inventory(item, qty)        write_to_warehouse(inventory_result)

# 创建一个部署，按cron调度github_flow.serve(    name="github-stats-daily",    cron="0 9 * * *",  # 每天早上9点运行    parameters={"repos": ["PrefectHQ/prefect", "pandas-dev/pandas"]})# 部署到EC2 Workerfrom prefect.deployments import Deployment, DockerImagedeployment = Deployment.build_from_flow(    flow=github_flow,    name="ecs-deployment",    work_pool_name="aws-ecs-pool",    image=DockerImage(name="my-registry/prefect-flows:latest"))deployment.apply()

from prefect import flow, taskfrom prefect.tasks import task_input_hashfrom datetime import timedeltaimport httpximport pandas as pd@task(    retries=3,    retry_delay_seconds=10,    cache_expiration=timedelta(hours=1),    cache_key_fn=task_input_hash)def extract_data(api_url: str) -> dict:    """从API提取数据，失败自动重试3次，相同输入会命中缓存"""    response = httpx.get(api_url, timeout=30.0)    response.raise_for_status()    return response.json()@task(log_prints=True)def transform_data(raw_data: dict) -> pd.DataFrame:    """清洗和转换数据"""    df = pd.DataFrame(raw_data.get("items", []))    df = df.drop_duplicates()    df["processed_at"] = pd.Timestamp.now()    print(f"Transformed {len(df)} records")    return df@taskdef load_data(df: pd.DataFrame, target_table: str) -> None:    """加载到目标表（示例中打印数据，实际可替换为数据库写入）"""    print(f"Loading {len(df)} rows to {target_table}")    # 实际生产环境中，这里会是 df.to_sql 或写入S3等操作@flowdef etl_pipeline(api_endpoints: list[str], target: str = "analytics.table"):    results = []    for endpoint in api_endpoints:        raw = extract_data(endpoint)        transformed = transform_data(raw)        results.append(transformed)    final_df = pd.concat(results, ignore_index=True)    load_data(final_df, target)    return final_dfif __name__ == "__main__":    # 本地运行    etl_pipeline(        ["https://api.example.com/users", "https://api.example.com/orders"],        target="staging.user_activity"    )    # 也可以部署为定时任务    # etl_pipeline.serve(name="hourly-etl", cron="0 * * * *")