Prefect

参考：Prefect

• Prefect 简介
  • Prefect 特性
    • 开源事件和自动化系统
    • 新的事务性接口
    • 灵活的任务执行
    • 增强的客户端引擎
    • 改进的工件和变量
    • 工作者
  • 性能提升
  • 安装 Prefect
• 教程
  • 快速上手 Prefect
    • 连接至Prefect API
    • 将您的脚本转换为 Prefect 工作流
    • 执行流程
  • 流程调度
  • 发布代码到远程仓库
  • 创建工作池
  • 部署并调度流程
  • 构建数据管道
    • 失败重试
    • 并行执行慢速任务
    • 避免达到速率限制
    • 缓存任务结果
    • 运行改善后的任务
  • 从网站提取数据
    • 设置错误处理机制
    • 摄取大量数据
    • 请根据依赖的嵌套流程和任务来结构化代码
    • 综合起来
• 开发手册
  • 编写与运行工作流程
    • 编写并运行流
      • 运行流
      • 指定流参数
      • 编排工作流
        • 嵌套流
      • 支持的功能
        • 异步函数
        • 类方法
        • 生成器
      • 流程运行
      • 流设置
      • 最终状态的确定
        • 返回未来值
        • 返回多个状态或者未来
        • 运行手动状态
          • 自定义命名状态
      • 重试
    • 编写和运行任务
      • 支持的函数
        • 同步函数
        • 异步函数
        • 类方法
        • 生成器
      • 并发性
      • 任务配置
      • 标签
      • 超时
      • 重试机制
        • 自定义重试行为
          • 增加“抖动”以避免雷鸣般的群体效应
          • 全局配置重试行为
    • 在后台运行任务
      • 使用延迟任务
        • 定义任务
        • 调用任务
        • 使用任务工作者执行延迟任务
      • 探索Prefect中的延迟任务和任务工作者
        • 设置
          • 第一步：激活虚拟环境
          • 第二步：安装Python依赖项
          • 第三步：连接到Prefect云或自托管的Prefect服务器
          • 步骤4：克隆仓库（可选）
        • 示例1：通过调用在前台运行Prefect任务
          • 步骤1：创建带有任务装饰的函数的文件
          • 第二步：在终端运行脚本
          • 可选操作
        • 示例2：启动任务工作者并在后台运行延迟任务
          • 步骤1：在文件中定义任务和任务工作者
        • 步骤2：通过在终端运行脚本启动任务工作者
        • 步骤3：创建文件并保存以下代码：
        • 步骤4：打开另一个终端并运行脚本
        • 步骤5：在 UI 中查看任务运行
        • 步骤6：使用多个任务工作者并行运行任务
        • 步骤7：向任务工作者发送多个任务
          • 步骤8：使用 control + c 关闭任务工作者
    • 可视化流程结构
    • 处理大数据
      • 移除任务内省
      • 将任务结果写入云存储
      • 在流程中保存数据到磁盘
      • 缓存任务结果
      • 压缩写入磁盘的结果
      • 使用任务运行器进行可并行操作
    • 执行以下 shell 命令
      • watch命令
        • 示例用法
        • watch的好处
      • 使用serve部署
        • 示例用法
        • serve的好处
    • 访问第三方密钥
    • 前提条件
    • 步骤
      • 步骤1：安装prefect-aws和prefect-snowflake库
      • 第二步：在AWS Secrets Manager中存储Snowflake密码
      • 第三步：创建AwsSecret块以访问你的Snowflake密码
      • 步骤4：创建AwsCredentials区块
      • 步骤5：确保计算环境能够访问AWS凭证
      • 步骤6：在Python代码中创建并使用SnowflakeCredentials和SnowflakeConnector块
    • 下一步
  • 配置运行时行为
    • 配置日志记录
      • Prefect日志记录器
        • 记录流程和任务
        • 记录打印语句
      • 日志配置
        • 自定义日志配置
        • 格式化器
        • 样式
        • 高亮日志
      • 从命令行访问日志
    • 配置任务缓存
      • 缓存键
      • 缓存策略
      • 自定义缓存
        • 缓存过期
        • 缓存策略
        • 缓存键函数
        • 缓存存储
        • 缓存隔离
        • 处理不可序列化对象
      • 多任务缓存
      • 缓存示例
      • 强制忽略缓存
    • 写事务
      • 编写你的第一个事务
      • 事务周期
      • 幂等性
        • 处理竞态条件
      • 在事务中访问数据
    • 暂停和恢复流程运行
      • 暂停流程运行
        • 挂起流程运行
      • 在暂停或挂起流程运行时等待输入
    • 取消流程运行
      • 通过命令行接口（CLI）取消流程运行
      • 通过用户界面（UI）取消
      • 超时设置
    • 发送和接收流程运行输入
      • 暂停或中断流程，直到它接收到输入
        • 等待输入
        • 提供初始数据
        • 提供带有运行时数据的描述
        • 处理自定义验证
      • 运行时发送和接收输入
        • 接收输入
        • 跟踪您已经看到的输入
        • 回应输入的发送者
        • 发送输入
        • 完整示例
    • 访问运行时上下文
      • 访问运行时信息
      • 直接访问运行上下文
    • 在 Python 中管理运行元数据
      • 示例
        • 重新安排延迟的流程运行
        • 从你的工作空间获取最后 N 个已完成的流程运行
        • 通过客户端将所有正在运行的流程转换为已取消状态
  • 管理状态和配置
    • Manage states
      • State types
      • Final state determination
        • Task return values
        • Flow return values
      • Execute code on state changes
        • Available state change hooks
        • Pass kwargs to state change hooks
        • Example usage: send a notification when a flow run fails
    • Manage results
      • Configuring result persistence
        • Default persistence configuration
        • Enabling result persistence
        • Result storage
          • Specifying a default filesystem
          • Result filenames
        • Result serialization
      • Advanced: Caching results in memory
    • Create run artifacts
      • Create artifacts
        • Create link artifacts
        • Create progress artifacts
        • Create Markdown artifacts
        • Create table artifacts
        • Create image artifacts
      • Manage artifacts
        • Reading artifacts
        • Fetching artifacts
        • Delete artifacts
      • Artifacts API
    • Set and get variables
      • Manage variables
        • Through the Prefect UI
        • Through the REST API
        • Through the CLI
      • Access variables
        • In Python code
        • In prefect.yaml deployment steps
    • Securely store typed configuration
      • How blocks work
        • Block type
        • Block document
        • Block class
      • Using blocks
        • Nested blocks
      • Create a new block type
        • Register custom blocks
        • Secret fields
        • Customize a block’s display
        • Update custom Block types
      • Pre-registered blocks
        • Built-in blocks
        • Blocks in Prefect integration libraries
  • 管理并发
    • Run tasks concurrently or in parallel
      • Configure a task runner
      • Access results from submitted tasks
      • Mapping over iterables
      • Use multiple task runners
    • Limit concurrent task runs with tags
      • Execution behavior
      • Configure concurrency limits
        • CLI
        • Python client
    • Apply global concurrency and rate limits
      • Manage global concurrency and rate limits
        • Active vs. inactive limits
        • Slot decay
        • Through the UI
        • Through the CLI
      • Using the concurrency context manager
      • Using rate_limit
      • Use concurrency and rate_limit outside of a flow
      • Use cases
        • Throttling task submission
        • Manage database connections
        • Parallel data processing
      • Other ways to limit concurrency
  • 管理设置
    • Configure settings and profiles
      • Why use settings?
      • Get started with settings
      • Settings sources
        • Environment variables
        • .env file
        • prefect.toml file
        • pyproject.toml file
        • Profiles
          • Configure settings for the active profile
      • View current configuration
      • Common client settings
      • Common server settings
        • Security settings
• 部署
  • Deploy overview
    • Create a deployment
      • Create a deployment with serve
      • Create a work pool-based deployment
      • Create a deployment with deploy
      • Create a deployment with a YAML file
      • Work pools
      • Work pool-based deployment requirements
    • Run a deployment
      • Work pools that require a worker
      • Work pools that don’t require a worker
    • Static vs. dynamic infrastructure
      • Static infrastructure
      • Dynamic infrastructure
    • Deployment schema
      • Required defining data
      • Workflow scheduling and parametrization
      • Concurrency limiting
      • Metadata for bookkeeping
      • Worker-specific fields
  • Run flows in local processes
    • Serve a flow
    • Additional serve options
    • Serve multiple flows at once
    • Retrieve a flow from remote storage
    • If you need dynamic infrastructure
  • 基础设施概念
    • 通过工作池配置动态基础设施
      • Work pool configuration
      • Work pool types
      • View work pools
      • Work pool status
      • Pause and delete work pools
      • Manage concurrency
        • Base job template
      • Work queues
        • Queue priority
        • Queue concurrency limits
        • Precise control with priority and concurrency
      • Next steps
    • Learn about workers
      • Worker types
      • Worker options
      • Worker status
      • Worker logs
      • Worker details
      • Start a worker
      • Configure prefetch
      • Polling for work
      • Install policy
      • Additional resources
    • Retrieve code from storage
      • Deployment creation options
      • Git-based storage
      • Docker-based storage
        • Custom Docker image
      • Cloud-provider storage
      • Store code locally
      • Include or exclude files from storage
      • Update flow code
      • Flow code storage for deployments created with serve
    • Deploy flows with Python
      • When to consider flow.deploy over flow.serve
      • Prerequisites
      • Deploy a flow with flow.deploy
        • Trigger a run
      • Deploy with a schedule
      • Use remote code storage
      • Set default parameters
      • Set job variables
      • Deploy multiple flows
      • Additional resources
    • Define deployments with YAML
      • Deployment actions
      • The build action
      • The push action
      • The pull action
      • Utility steps
      • Templating options
      • Work with multiple deployments with prefect.yaml
      • Reuse configuration across deployments
      • Deployment declaration reference
        • Deployment fields
        • Schedule fields
        • Concurrency limit fields
        • Work pool fields
        • Deployment mechanics
      • Update a deployment
      • Next steps
    • Build deployments via CI/CD
      • Get started with GitHub Actions and Prefect
        • Repository secrets
        • Write a GitHub workflow
        • Run a GitHub workflow
      • Advanced example
        • Deploy to multiple workspaces
        • Caching build dependencies
      • Prefect GitHub Actions
      • Authenticate to other Docker image registries
      • See also
    • Override job configuration for specific deployments
      • Job variables
      • Override job variables on a deployment
        • Use a prefect.yaml file
          • Hard-coded job variables
          • Use existing environment variables
        • Use the .deploy() method
      • Override job variables on a flow run
        • Use the custom run form in the UI
        • Use the CLI
        • Use job variables in automations
  • 基础设施示例
    • Run flows in Docker containers
      • Create a work pool
      • Start a worker
      • Create the deployment
        • Automatically bake your code into a Docker image
        • Automatically build a custom Docker image with a local Dockerfile
        • Store your code in git-based cloud storage
        • Additional configuration with .deploy
        • Work with multiple deployments with deploy
      • Learn more
    • Run flows on Kubernetes
      • Prerequisites
      • Create a cluster
      • Create a container registry
      • Create a Kubernetes work pool
        • Configure work pool options
      • Create a Prefect Cloud API key
      • Deploy a worker using Helm
        • Add the Prefect Helm repository
        • Create a namespace
        • Create a Kubernetes secret for the Prefect API key
        • Configure Helm chart values
        • Create a Helm release
        • Verify deployment
      • Define a flow
      • Define a Prefect deployment
        • Tag images with a Git SHA
      • Authenticate to Prefect
      • Deploy the flows
      • Run the flows
    • Run flows on Prefect Managed infrastructure
      • Create a managed deployment
      • Add dependencies
      • Limitations
        • Concurrency and work pools
        • Images
        • Code storage
        • Resources
      • Usage limits
      • Next steps
    • Run flows on serverless compute
      • Automatic infrastructure provisioning
        • Prerequisites
        • Automatically create a new push work pool and provision infrastructure
        • Use existing resources with automatic infrastructure provisioning
      • Provision infrastructure for an existing push work pool
      • Manual infrastructure provisioning
        • Work pool configuration
        • Create a Credentials block
        • Create a push work pool
      • Deployment
      • Putting it all together
      • Next steps
  • 静态基础设施示例
    • Serve flows in a long-lived Docker container
      • Overview
      • Writing the flow
      • Writing the Dockerfile
      • Build and run the container
        • Build (and push) the image
        • Run the container
          • Verify the container is running
          • View logs
        • Stop the container
      • Next steps
  • Daemonize worker processes
    • Prerequisites
    • Steps
      • Step 1: Add a user
      • Step 2: Install Prefect
      • Step 3: Set up a systemd service
        • Setting up a systemd service for a Prefect worker
        • Setting up a systemd service for .serve
      • Step 4: Save, enable, and start the service
• 自动化概述
  • 调度流程运行
    • 创建时间表
      • 在 UI 中创建调度
      • 使用 Python 创建调度
      • 在终端中创建调度
      • 在 YAML 中创建调度
    • 调度类型
      • Cron
        • day_or 属性的工作原理
      • Interval
      • RRule
    • 调度的工作原理
  • 事件
    • 通过事件跟踪活动
      • 事件规范
      • 事件语法
      • 事件源
        • 从 Python 代码发出自定义事件
        • 通过 Webhook 发出事件
      • 资源
      • UI 中的事件
      • 响应事件
    • 在事件上触发操作
      • 创建自动化
      • 管理自动化
        • 触发器
        • 操作
        • 在 Python 代码中创建自动化
        • 选定和推断的操作目标
      • 使用部署触发器创建自动化
        • 在 prefect.yaml 中定义触发器
        • 在 .serve 和 .deploy 中定义触发器
        • 将触发器传递给 prefect deploy
      • 使用自动化发送通知
      • 使用 Jinja 模板
      • API 示例
        • 创建示例脚本
        • 在 UI 中创建通知块
        • 基于事件的部署自动化
      • 使用底层的 .yaml 文件
      • 使用自定义 Webhook 启动自动化
      • 示例
        • 使用事件触发下游部署
        • 当客户完成订单时触发部署
      • 另请参阅
    • Define custom event triggers
      • Event triggers
      • Resource matching
      • Expected events
      • Evaluation strategy
      • Metric triggers
      • Composite triggers
    • Receive events with webhooks
      • Configure webhooks
        • Through the Prefect Cloud API
        • Through Prefect Cloud
        • Through the Prefect CLI
      • Webhook endpoints
      • Webhook authentication
        • API key authentication
        • Account-level security enforcement
      • Webhook templates
        • Static webhook events
        • Event fields that Prefect Cloud populates for you
        • Dynamic webhook events
        • How HTTP request components are handled
        • Accept Prefect events directly
        • Accepting CloudEvents
      • Troubleshooting