prefecthq/prefect-azure

Prefect Azure集成镜像

镜像概述

Prefect Azure集成镜像是专为在Azure云环境中使用Prefect工作流编排工具设计的专用镜像。该镜像基于Prefect基础镜像构建，预装了与Azure服务集成所需的依赖组件，旨在简化用户在Azure平台上构建、调度和监控工作流的过程。

核心功能与特性

• Azure服务集成：原生支持与Azure关键服务的对接，包括Azure Blob Storage（对象存储）、Azure Container Instances（容器实例）、Azure SQL Database（关系型数据库）等
• 预装依赖组件：包含azure-storage-blob、azure-mgmt-containerinstance、azure-identity等必要Python库，无需额外安装
• Prefect工作流支持：兼容Prefect Flow定义、任务调度、状态监控等核心功能，可直接运行Prefect工作流脚本
• 配置简化：通过环境变量即可配置Azure认证信息（如服务主体、连接字符串）和Prefect Cloud连接参数

使用场景与适用范围

• 数据工程工作流：数据工程师在Azure环境中构建ETL/ELT流程，实现数据从Azure Blob Storage到Azure SQL的传输与处理
• 云资源自动化：DevOps团队通过Prefect工作流自动化Azure资源（如虚拟机、容器实例）的创建、配置与扩展
• 科研计算任务：研究团队在Azure上运行大规模数据处理或机器学习训练工作流，利用Prefect进行任务依赖管理和失败重试
• 跨服务协调：需要在多个Azure服务间（如Azure Functions、Azure Data Lake、Azure Event Hub）协调执行的复杂业务流程

使用方法与配置说明

基本运行（Docker Run）

bash
docker run -it --rm \
  -e PREFECT_API_KEY=<your-prefect-api-key> \
  -e AZURE_STORAGE_CONNECTION_STRING=<your-azure-storage-connection-string> \
  -v ./your-flows:/opt/prefect/flows \
  prefecthq/prefect-azure:latest

Docker Compose配置示例

yaml
version: '3.8'
services:
  prefect-azure-agent:
    image: prefecthq/prefect-azure:latest
    restart: always
    environment:
      - PREFECT_API_KEY=pnu_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
      - PREFECT_API_URL=https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>
      - AZURE_CLIENT_ID=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
      - AZURE_TENANT_ID=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
      - AZURE_CLIENT_SECRET=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
    volumes:
      - ./flows:/opt/prefect/flows
    command: prefect agent start -q azure-agent-queue

核心环境变量配置

工作流示例（Python）

python
from prefect import flow, task
from prefect_azure import AzureBlobStorageCredentials
from prefect_azure.blob_storage import blob_storage_upload

@task
def upload_to_azure_blob():
    credentials = AzureBlobStorageCredentials.from_connection_string(
        connection_string=os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    )
    blob_storage_upload(
        data=b"sample data",
        blob_name="sample.txt",
        container="my-container",
        credentials=credentials
    )

@flow
def azure_integration_flow():
    upload_to_azure_blob()

if __name__ == "__main__":
    azure_integration_flow()

将上述脚本保存为./flows/azure_flow.py，通过Docker Compose启动容器后，即可在Prefect Cloud中调度执行该工作流。