prefect-azure 集成实战指南：Blob Storage、容器实例 Worker 与数据库托管身份认证-迪斯科星球

prefect-azure 集成实战指南：Blob Storage、容器实例 Worker 与数据库托管身份认证

【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect

prefect-azure是 Prefect 官方的 Azure 集成集合，用于在 Prefect 工作流中编排各类 Azure 服务：读写 Blob Storage 对象、查询 Cosmos DB、对接 Azure ML Datastore、把 flow 运行部署到 Azure Container Instances（ACI），并为 Prefect server 提供基于 Microsoft Entra ID（托管身份）的免密数据库认证。读完本文，你将掌握prefect-azure的安装与块注册、多种 Azure 认证方式的取舍、Blob Storage 任务的用法，以及如何搭建一套完整的 ACI 混合工作池并启用 Postgres 托管身份认证插件。

集成概览：prefect-azure 能做什么

prefect-azure是一个遵循 Prefect 集合（collection）规范的独立 Python 包，通过 pyproject.toml 声明了prefect.collections与prefect.plugins两类入口点，前者让 Prefect 能自动发现并注册其中的块类型，后者则用于加载数据库认证插件。包的核心能力集中在 prefect_azure 包中：

五类凭据块：AzureBlobStorageCredentials、AzureCosmosDbCredentials、AzureMlCredentials、AzureContainerInstanceCredentials、AzureDevopsCredentials；
Blob Storage 任务与容器块：blob_storage_download、blob_storage_upload、blob_storage_list以及AzureBlobStorageContainer块；
Cosmos DB 查询任务：cosmos_db_query_items、cosmos_db_read_item等；
Azure ML Datastore 任务：列出、获取与上传 Datastore 文件；
ACI Worker：AzureContainerWorker，以混合工作池模式在 Azure 容器实例中运行 flow；
部署步骤：prefect_azure.deployments.steps提供 Blob Storage 的 push/pull 步骤，供prefect.yaml使用；
插件：为 Prefect server 的 Postgres 数据库接入 Entra ID 托管身份认证。

安装与块注册

基础安装

prefect-azure支持 Python 3.10 及以上版本，依赖prefect>=3.8.4、prefect-docker>=0.6.2、azure_identity、azure_mgmt_containerinstance、azure-mgmt-resource、aiohttp等（见 pyproject.toml 的dependencies段）。推荐使用与 Prefect 版本自动匹配的安装方式：

pip install "prefect[azure]"

升级到最新版本：

pip install -U "prefect[azure]"

不同功能对应不同的可选依赖（extras），按需安装：

# 使用 Blob Storage 功能 pip install "prefect-azure[blob_storage]" # 使用 Cosmos DB 功能 pip install "prefect-azure[cosmos_db]" # 使用 ML Datastore 功能（依赖 azureml-core） pip install "prefect-azure[ml_datastore]" # 一次性安装全部可选依赖 pip install "prefect-azure[all_extras]"

从 pyproject.toml 的[project.optional-dependencies]段可以看到，这些 extras 分别对应azure-storage-blob、azure-cosmos、azureml-core。需要说明的是，若未安装对应的 Azure SDK，相关任务在调用时会被_raise_help_msg装饰器拦截并抛出包含安装指引的ImportError，提示你补装对应 extra（见 credentials.py）。

注册块类型

安装完成后，需要让 Prefect 注册新块类型：

prefect block register -m prefect_azure

注册后即可在 Prefect UI 或通过Block.load("块名")使用这些凭据块。

认证机制：五类凭据块

AzureBlobStorageCredentials：三种认证方式

AzureBlobStorageCredentials是使用最频繁的凭据块，支持三种认证方式（见 credentials.py 中的check_connection_string_or_account_url校验逻辑）：

连接字符串：只提供connection_string，直接用于BlobServiceClient.from_connection_string；
账户 URL + DefaultAzureCredential：只提供account_url，通过DefaultAzureCredential自动发现环境凭据（本地az login、环境变量、托管身份等）；
账户 URL + 服务主体（SPN）：同时提供account_url、client_id、tenant_id、client_secret，使用ClientSecretCredential。

块的字段模型约束如下：

字段	是否必填	说明
`connection_string`	二者必居其一	存储账户连接字符串，优先级高于`account_url`，不能与 SPN 字段混用
`account_url`	二者必居其一	存储账户 URL，如`https://myaccount.blob.core.windows.net/`
`client_id`/`tenant_id`/`client_secret`	SPN 模式必填	三者要么全填要么全不填，且要求提供`account_url`

源码中的model_validator会严格执行以下规则：SPN 字段不全时报错；有 SPN 无account_url时报错；连接字符串与 SPN 混用时报错；connection_string与account_url同时提供时报错。

其他凭据块

AzureCosmosDbCredentials：仅需connection_string，提供get_client()、get_database_client(database)、get_container_client(container, database)用于 Cosmos DB 认证；
AzureMlCredentials：需要tenant_id、service_principal_id、service_principal_password、subscription_id、resource_group、workspace_name六项，通过get_workspace()返回 Azure MLWorkspace对象；
AzureContainerInstanceCredentials：服务主体三件套（client_id/tenant_id/client_secret）均提供时使用ClientSecretCredential，否则回退到DefaultAzureCredential；支持credential_kwargs透传额外参数；get_container_client(subscription_id)返回ContainerInstanceManagementClient；
AzureDevopsCredentials：保存个人访问令牌（PAT），get_auth_header()生成 Basic Auth 请求头，可用于 Azure DevOps REST API。

Blob Storage：对象读写任务

下载对象

blob_storage.py 中的blob_storage_download是一个@task装饰的异步任务，接收容器名、blob 名与凭据，返回bytes：

from prefect import flow from prefect_azure import AzureBlobStorageCredentials from prefect_azure.blob_storage import blob_storage_download @flow def example_blob_storage_download_flow(): connection_string = "connection_string" blob_storage_credentials = AzureBlobStorageCredentials( connection_string=connection_string, ) data = blob_storage_download( blob="prefect.txt", container="prefect", azure_credentials=blob_storage_credentials, ) return data example_blob_storage_download_flow()

从实现看，该任务内部通过async with blob_storage_credentials进入异步上下文（自动管理凭据生命周期），再经由get_blob_client(container, blob)调用download_blob()并读取为字节内容，最后在__aexit__中关闭凭据资源。任务中还会通过get_run_logger()输出下载日志，便于在 Prefect 运行日志中追踪。

上传与列出

blob_storage_upload支持overwrite参数控制是否覆盖已存在的 blob（默认False，blob 已存在时抛错；overwrite=True则替换），未指定blob名时会自动生成一个 UUID 作为键名：

from prefect import flow from prefect_azure import AzureBlobStorageCredentials from prefect_azure.blob_storage import blob_storage_upload @flow def example_blob_storage_upload_flow(): connection_string = "connection_string" blob_storage_credentials = AzureBlobStorageCredentials( connection_string=connection_string, ) with open("data.csv", "rb") as f: blob = blob_storage_upload( data=f.read(), container="container", blob="data.csv", blob_storage_credentials=blob_storage_credentials, overwrite=False, ) return blob example_blob_storage_upload_flow()

blob_storage_list用于列出容器内的对象，支持name_starts_with前缀过滤与include参数（可选值包括snapshots、metadata、uncommittedblobs、copy、deleted、tags、versions等），其余关键字参数透传给 Azure SDK 的ContainerClient.list_blobs()。

用 with_options 定制任务行为

任何 Prefect task 或 flow 都可以通过with_options生成定制副本，例如重命名、设置重试策略：

custom_blob_storage_download_flow = example_blob_storage_download_flow.with_options( name="My custom task name", retries=2, retry_delay_seconds=10, )

AzureBlobStorageContainer 块：对象存储文件系统

除了函数式任务，prefect-azure还提供AzureBlobStorageContainer块，它同时继承了 Prefect 的ObjectStorageBlock、WritableFileSystem与WritableDeploymentStorage抽象接口，字段包括container_name、credentials和可选的base_folder（容器内基准目录，读写路径会相对它解析）。

该块的方法覆盖了对象存储与文件系统操作：

下载：download_object_to_path、download_object_to_file_object、download_folder_to_path（递归下载目录）、read_path（返回字节，支撑结果存储）；
上传：upload_from_path、upload_from_file_object、upload_from_folder、write_path（以overwrite=True写入，保证任务refresh_cache=True重跑时能覆盖旧 blob）、put_directory（支持ignore_file忽略规则，支撑部署代码存储）；
列举：list_blobs、get_directory。

这些方法均以@sync_compatible装饰，同步调用和await异步调用皆可，方便在不同运行环境中复用。put_directory/get_directory的实现使其可作为部署的流代码存储后端，read_path/write_path则使其可作为 flow 运行结果存储后端。

部署步骤：把 flow 代码推送/拉取到 Blob Storage

prefect_azure.deployments.steps提供了两个部署步骤，可在prefect.yaml的push与pull阶段引用，实现"代码进 Blob、运行前再拉取"的部署形态。结合服务主体认证示例：

push: - prefect_azure.deployments.steps.push_to_azure_blob_storage: requires: prefect-azure[blob_storage] container: my-container folder: my-folder credentials: "{{ prefect.blocks.azure-blob-storage-credentials.my-spn-credentials }}" pull: - prefect_azure.deployments.steps.pull_from_azure_blob_storage: requires: prefect-azure[blob_storage] container: "{{ container }}" folder: "{{ folder }}" credentials: "{{ prefect.blocks.azure-blob-storage-credentials.my-spn-credentials }}"

其中credentials通过{{ prefect.blocks.azure-blob-storage-credentials.<块名> }}模板语法引用已保存的凭据块。SPN 凭据块可以这样创建并保存：

from prefect_azure import AzureBlobStorageCredentials credentials = AzureBlobStorageCredentials( account_url="https://myaccount.blob.core.windows.net/", tenant_id="your-tenant-id", client_id="your-client-id", client_secret="your-client-secret", ) credentials.save("my-spn-credentials")

Cosmos DB 与 ML Datastore

Cosmos DB 查询

prefect_azure.cosmos_db提供cosmos_db_query_items、cosmos_db_read_item、cosmos_db_create_item、cosmos_db_upsert_item、cosmos_db_delete_item等任务，支持传入 SQL 查询与参数化查询。示例：

from prefect import flow from prefect_azure import AzureCosmosDbCredentials from prefect_azure.cosmos_db import cosmos_db_query_items @flow def example_cosmos_db_query_items_flow(): connection_string = "connection_string" cosmos_db_credentials = AzureCosmosDbCredentials(connection_string) query = "SELECT * FROM c where c.age >= @age" container = "Persons" database = "SampleDB" parameters = [dict(name="@age", value=44)] results = cosmos_db_query_items( query, container, database, cosmos_db_credentials, parameters=parameters, enable_cross_partition_query=True, ) return results example_cosmos_db_query_items_flow()

从 cosmos_db.py 的实现看，这类任务是典型的"同步 SDK 异步化"模式：通过anyio的to_thread.run_sync把同步的 Cosmos SDK 调用放到线程中执行，避免阻塞事件循环。

ML Datastore

prefect_azure.ml_datastore提供ml_list_datastores、ml_get_datastore、ml_upload_file、ml_upload_folder、ml_download_file、ml_download_folder等任务，基于AzureMlCredentials.get_workspace()返回的 Azure ML Workspace 操作 Datastore 数据引用。

Azure Container Instance Worker：在 ACI 中运行 flow

ACI Worker 允许把 flow 运行提交到 Azure Container Instances，实现按需伸缩的混合工作池部署。快速上手只需两步：

创建 ACI 类型的工作池

prefect work-pool create --type azure-container-instance my-aci-work-pool

启动 Worker

prefect worker start --pool my-aci-work-pool --type azure-container-instance

Worker 会轮询工作池中的调度任务，并为每个 flow run 在 ACI 中创建容器组执行。AzureContainerWorker的实现位于 workers/container_instance.py，其关键设计包括：

ARM 模板驱动：默认使用一段 ARM（Azure Resource Manager）模板创建容器组，模板中的{{ command }}、{{ cpu }}、{{ memory }}等占位符在运行时替换为具体值；模板与变量均可按工作池在 Prefect UI 中定制，使 worker 不受内置功能限制；
资源默认值：CPU 默认1.0、内存默认1.0GB、GPU 默认0.0；
安全变量：PREFECT_API_KEY、PREFECT_API_AUTH_STRING会被 ACI 视为安全环境变量，不会出现在日志中；
健壮性：容器组删除设置了 30 秒超时以避免遗留孤儿资源；对 Azure ARM API 的瞬时错误（如 HTTP 503）设置了最多 3 次重试与退避；
ACR 拉取：通过ACRManagedIdentity或DockerRegistryCredentials（来自 prefect-docker）认证，从 Azure Container Registry 拉取 flow 镜像。

从零搭建 ACI 混合工作池的完整流程

更完整的 ACI 部署路径参见 aci_worker.mdx，核心步骤如下：

创建资源组并记录其 scope：
```
export RG_NAME=<resource-group-name> && \ az group create --name $RG_NAME --location <location> RG_SCOPE=$(az group show --name $RG_NAME --query id --output tsv)
```
RG_SCOPE形如/subscriptions/<subscription-id>/resourceGroups/<resource-group-name>，后续角色与身份配置都要用到。
准备 ACI 权限：创建自定义角色Container Instances Contributor（包含Microsoft.ManagedIdentity/userAssignedIdentities/assign/action、Microsoft.Resources/deployments/*、Microsoft.ContainerInstance/containerGroups/*三项权限），创建用户托管身份，并把该角色以及内置AcrPull角色分配给身份。
创建 worker 容器实例：用az container create在 Azure 中启动常驻 worker，镜像使用prefecthq/prefect-azure:latest（已预装 prefect 与 prefect-azure；生产环境建议固定版本标签），并通过--secure-environment-variables传入PREFECT_API_URL与PREFECT_API_KEY。注意 API URL 必须能被 ACI 容器访问，不能使用127.0.0.1之类的本机地址：
```
az container create \ --name <work-pool-name> \ --resource-group $RG_NAME \ --assign-identity $IDENTITY_ID \ --image "prefecthq/prefect-azure:latest" \ --secure-environment-variables PREFECT_API_URL=$PREFECT_API_URL PREFECT_API_KEY=$PREFECT_API_KEY \ --command-line "/bin/bash -c 'prefect worker start --pool <work-pool-name> --type azure-container-instance'"
```
创建 ACR 注册表：az acr create创建 Azure Container Registry，用于存放包含 flow 代码的 Docker 镜像。
更新工作池配置：在 Prefect UI 的工作池编辑页中，把IDENTITY_ID填入Identities（JSON 数组格式），配置ACRManagedIdentity（身份 ID 与<registry-name>.azurecr.io），并填入订阅 ID 与资源组名。
部署并运行 flow：在本地写 flow 并用my_flow.deploy(...)（Python 方式）或prefect.yaml（build/push/pull 步骤 +prefect deploy --all）创建部署，随后用prefect deployment run "my-flow/aci-deployment"触发；worker 会从 ACR 拉取镜像并为 flow run 创建新的容器实例。

Prefect server 数据库：Entra ID 托管身份免密认证

prefect-azure内置一个 Prefect 插件，让Prefect server以 Microsoft Entra ID（托管身份）令牌而非密码连接 Azure Database for PostgreSQL。其实现位于 plugins.py：通过@register_hook注册set_database_connection_params钩子，Prefect 在构建 server 数据库引擎时会调用它并把返回的映射合并进asyncpg的connect_args。

核心机制：

短期令牌自动续期：钩子返回的password是一个可调用对象，asyncpg在每条新连接建立时都会调用它，通过DefaultAzureCredential获取https://ossrdbms-aad.database.windows.net/.default作用域下的令牌；azure-identity会缓存令牌，临近过期时自动刷新；
凭据单例复用：DefaultAzureCredential在每个引擎生命周期内只实例化一次，内部令牌缓存跨连接复用，避免每条连接新建凭据导致令牌端点被节流；
强制 SSL：Entra 认证要求 SSL，钩子会用ssl.create_default_context()（开启证书校验的安全默认值）配置连接；
设置驱动：由 settings.py 中的AzureSettings.postgres.managed_identity控制，enabled默认False；关闭时钩子返回空映射，不影响正常密码认证。

启用步骤

在运行prefect server start的进程环境（镜像）中安装prefect-azure，然后配置环境变量：

# 启用插件系统（Prefect < 3.7 使用 PREFECT_EXPERIMENTS_PLUGINS_ENABLED=true） export PREFECT_PLUGINS_ENABLED=true # 开启托管身份认证 export PREFECT_INTEGRATIONS_AZURE_POSTGRES_MANAGED_IDENTITY_ENABLED=true # 可选：指定用户分配的托管身份 client_id export PREFECT_INTEGRATIONS_AZURE_POSTGRES_MANAGED_IDENTITY_CLIENT_ID=<client-id> # 提供无密码连接 URL（令牌由插件提供） export PREFECT_SERVER_DATABASE_CONNECTION_URL="postgresql+asyncpg://<entra-principal>@<host>:5432/<db>"

Azure 侧的前置条件

开启 Entra 认证：在 Azure Database for PostgreSQL 灵活服务器上启用 Microsoft Entra 身份验证；

创建数据库主体：以 Entra 管理员身份连接后，使用pgaadauth扩展为身份创建数据库角色并授权：

SELECT * FROM pgaadauth_create_principal('<identity-name>', false, false); GRANT CONNECT ON DATABASE prefect TO "<identity-name>";

提供身份：在 Azure 内运行时使用用户分配或系统分配的托管身份；本地开发时DefaultAzureCredential会自动回退到az login的开发者身份，因此同一套配置可直接用于开发环境。

启用后，数据库连接不再依赖任何持久化的明文密码，令牌在每次新连接时自动获取与刷新，降低了凭据泄露风险，也免去了密码轮换的运维负担。相关验证可参考 test_plugins.py。

小结

prefect-azure覆盖了 Prefect 与 Azure 集成的主要场景：五类凭据块统一了连接字符串、DefaultAzureCredential 与服务主体三种认证方式；Blob Storage 任务与AzureBlobStorageContainer块既支持函数式调用，也支持作为部署代码存储与结果存储后端；Cosmos DB 与 ML Datastore 任务补齐了数据层能力；ACI Worker 提供完整的混合工作池部署路径；而数据库托管身份认证插件则为 Prefect server 带来了免密、自动续期、无明文凭据存储的 Postgres 连接方案。无论你是在 Azure 上运行数据管道、把 flow 容器化部署到 ACI，还是想加固 Prefect server 的数据库访问，都可以从上述配置直接入手。

【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析