2026/9/14 3:54:46
FEATURED · 精选文章 
Apache Airflow Common Messaging用 MessageQueueTrigger 实现基于消息队列的事件驱动 DAG【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAirflow 的 Common Messaging Provider 提供了一个统一入口MessageQueueTrigger让你无需关心底层队列AWS SQS、Kafka、Redis pub/sub、Azure Service Bus 等的具体实现细节就能把队列里来了新消息变成触发 DAG 运行的事件。本文基于官方文档 triggers.rst 完整展开先给出 SQS 场景的可复制示例与消息 payload 的获取方法再深入源码 msg_queue.py 讲解 scheme 匹配、provider 注册与事件透传机制帮你把定时调度升级为事件驱动。一、核心概念Trigger、Asset 与 AssetWatcher 如何协作在 Airflow 中一个监听外部队列的 DAG 由三要素组成Message Queue TriggerMessageQueueTrigger负责从外部消息队列如 AWS SQS、Kafka 或其他消息系统监听消息。它本身是一个统一封装——内部会根据你传入的scheme动态选择具体的 provider 触发器。Asset 与 AssetWatcherAsset抽象了外部实体例如那条 SQS 队列AssetWatcher则把一个具名 trigger挂到 asset 上。这个 name 用于标识哪个 trigger 对应哪个 asset是排查问题时的关键索引。事件驱动 DAGDAG 不再按固定周期运行而是当 asset 收到更新即队列里出现了新消息时执行。这种模式把调度权从时钟交给了消息适合下游数据处理、告警响应等对时效敏感、又不想高频空轮询的场景。二、等待队列消息SQS 完整示例官方文档给出的最小配置方式使用MessageQueueTrigger监听队列必填参数是scheme队列方案例如kafka、redispubsub、sqs。不同队列 provider 还支持各自的附加参数同时需要预先配置好连接使用相应的默认连接 ID——例如连接 AWS SQS 时连接 ID 应为aws_default。下面是系统测试中使用的真实示例 DAG位于 example_message_queue_trigger.pyfrom airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger from airflow.providers.standard.operators.empty import EmptyOperator from airflow.sdk import DAG, Asset, AssetWatcher # Define a trigger that listens to an external message queue (AWS SQS in this case) trigger MessageQueueTrigger(schemesqs, sqs_queuehttps://sqs.us-east-1.amazonaws.com/0123456789/Test) # Define an asset that watches for messages on the queue asset Asset(sqs_queue_asset, watchers[AssetWatcher(namesqs_watcher, triggertrigger)]) with DAG(dag_idexample_msgq_watcher, schedule[asset]) as dag: EmptyOperator(task_idtask)要点拆解schemesqs声明队列方案。sqs由 Amazon Provider 注册的SqsMessageQueueProvider认领其trigger_kwargs会把sqs_queue等参数透传给底层的SqsSensorTrigger见 sqs.py。sqs_queueSQS 队列的完整 URL作为 keyword argument 直接传给底层 trigger。Asset(..., watchers[AssetWatcher(name..., trigger...)])watcher 名称如sqs_watcher用于标识 trigger 与 asset 的对应关系。schedule[asset]把 asset 作为调度条件DAG 在该 asset 收到事件时触发运行。三、在 DAG 任务中使用消息 payload这是文档中最具实战价值的一段当消息到达时trigger 会把消息 payload 作为 trigger event 的一部分传递出来你可以在任务里通过triggering_asset_events参数直接读取from airflow.decorators import task from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger from airflow.sdk import DAG, Asset, AssetWatcher, chain # Define the asset with trigger trigger MessageQueueTrigger(schemekafka, topics[my-kafka-topic]) asset Asset(kafka_queue_asset, watchers[AssetWatcher(namekafka_watcher, triggertrigger)]) with DAG(dag_idexample_msgq_payload, schedule[asset]) as dag: task def process_message(triggering_asset_events): for event in triggering_asset_events[asset]: # Access the message payload payload event.extra[payload] # Process the payload as needed print(fReceived message: {payload}) chain(process_message())说明triggering_asset_events参数包含触发了本次 DAG 运行的事件按 asset 索引。每个 event 都带有extra字典消息 payload 存放在payload键下。例中使用schemekafka与topics[my-kafka-topic]Kafka 侧由KafkaMessageQueueProvider注册见 kafka/provider.yaml 中queues字段声明。这意味着下游任务可以直接拿到消息内容做业务处理而不是仅知道队列有新消息了。四、参数详解scheme、queue已废弃与 trigger_queue结合源码 msg_queue.pyMessageQueueTrigger.__init__接受以下参数参数必填说明scheme二选一队列方案如kafka、redispubsub、sqs用于 provider 匹配是推荐用法queue二选一已废弃。队列标识符URI 格式。若同时提供优先级高于scheme使用时会抛出AirflowProviderDeprecationWarningtrigger_queue否把该 trigger 分配到某个 triggerer 命名队列对应配置triggerer.queues_enabled与airflow triggerer --queues选项**kwargs否透传给底层 provider trigger 的参数如sqs_queue、topics、aws_conn_id等约束与行为queue与scheme必须提供其一否则抛出ValueError(Eitherqueueorschemeparameter must be provided.)。传入queue会触发废弃警告官方建议改用scheme并把配置以 keyword arguments 形式传入。trigger_queue与废弃的queue参数刻意命名区分因为它用于 triggerer 路由而非 broker 队列 URI重命名可避免破坏既有调用。trigger_queue的实际价值当不同队列 trigger 的资源消耗差异很大时可以用airflow triggerer --queues把重的监听任务隔离到独立队列避免相互饿死。五、源码剖析统一触发器如何路由到具体 providerMessageQueueTrigger的核心在trigger这个cached_propertymsg_queue.py其工作流程为动态发现模块加载时通过ProvidersManager().initialize_providers_queues()读取所有已安装 provider 的provider.yaml中声明的队列类实例化后放入MESSAGE_QUEUE_PROVIDERS。因此新 provider 只需在自己的provider.yaml中注册队列类即可被自动发现无需改动本模块。匹配若提供了queue旧方式逐个调用 provider 的queue_matches(queue_uri)做 URI 匹配如 SQS 的正则^https://sqs\.[^.]\.amazonaws\.com(\.cn)?/[0-9]/.定义于 sqs.py若提供scheme则调用scheme_matches(scheme)做精确字符串比较基类实现见 base_provider.py。冲突检测没有任何 provider 匹配时报ValueError并列出所有已注册 provider 名称多个 provider 同时匹配则报colliding错误。这种设计确保一个 scheme/queue 的归属唯一。构造底层 trigger匹配成功后取provider.trigger_class()生成真正的监听器实例如SqsSensorTrigger并把 kwargs 透传。事件透传run()方法只是async for event in self.trigger.run(): yield event即MessageQueueTrigger完全是个装饰器事件含extra[payload]原样从底层 trigger 流向 Asset 事件系统最终成为triggering_asset_events中的事件对象。从源码结构看serialize()直接委托给底层 trigger 的序列化因此 triggerer 重启恢复时行为与直接使用具体 trigger 完全一致。六、当前仓库中可用的队列 Provider 一览各 provider 在自己的provider.yaml的queues字段中注册队列类MessageQueueTrigger会自动发现它们。本仓库中当前注册的有Provider队列类schemeAmazonairflow.providers.amazon.aws.queues.sqs.SqsMessageQueueProvidersqsApache Kafkaairflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProviderkafkaGoogleairflow.providers.google.event_scheduling.events.pubsub.PubSubMessageQueueEventTriggerContainerpub/sub 相关IBM MQairflow.providers.ibm.mq.queues.mq.IBMMQMessageQueueProvideribm mq 相关Microsoft Azureairflow.providers.microsoft.azure.queues.asb.AzureServiceBusMessageBusMessageQueueProviderService Busazure service bus 相关Redisairflow.providers.redis.queues.redis.RedisPubSubMessageQueueProviderredispubsub具体注册行可见 amazon/provider.yaml、kafka/provider.yaml、redis/provider.yaml 等文件中的queues:段。可用的 scheme 集合取决于你安装了哪些 provider 包。七、运行前提与注意事项必须安装对应 provider 包MessageQueueTrigger本身不含任何队列实现若MESSAGE_QUEUE_PROVIDERS为空未安装任何队列类 providertrigger属性会抛出ValueError(No message queue providers are available.)并给出安装提示。连接配置各底层 trigger 依赖默认连接SQS 场景为aws_default。请先在 Airflow Connections 中配好凭据。需要 Triggererasset watcher 的 trigger 运行在 Triggerer 进程中事件驱动 DAG 要真正生效需确保 triggerer 已启动生产部署中如 Helm chart 等场景通常已包含该组件。废弃参数迁移如果你的旧代码使用了queuehttps://sqs....形式建议尽快迁移到schemesqs, sqs_queue...写法——源码表明queue优先级虽更高但已被标记废弃并将在未来版本移除。匹配唯一性不同 provider 的scheme_matches/queue_matches必须互不重叠一旦冲突会在创建 trigger 时立即报错这为多 provider 共存提供了保护。小结MessageQueueTrigger用一个 trigger、一个 scheme 参数抽象掉了 SQS、Kafka、Redis pub/sub 等多种消息系统的差异配合AssetAssetWatcherschedule[asset]的组合你可以在不写任何 sensor 长轮询代码的情况下构建出消息到达即触发的 DAG并通过triggering_asset_events[asset]中事件的extra[payload]直接消费消息内容。相关实现集中在 providers/common/messaging 目录单元与系统测试分别在 test_msg_queue.py 和 example_message_queue_trigger.py可作为行为参考。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Apache Airflow Common Messaging：用 MessageQueueTrigger 实现基于消息队列的事件驱动 DAG
Apache Airflow Common Messaging用 MessageQueueTrigger 实现基于消息队列的事件驱动 DAG【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAirflow 的 Common Messaging Provider 提供了一个统一入口MessageQueueTrigger让你无需关心底层队列AWS SQS、Kafka、Redis pub/sub、Azure Service Bus 等的具体实现细节就能把队列里来了新消息变成触发 DAG 运行的事件。本文基于官方文档 triggers.rst 完整展开先给出 SQS 场景的可复制示例与消息 payload 的获取方法再深入源码 msg_queue.py 讲解 scheme 匹配、provider 注册与事件透传机制帮你把定时调度升级为事件驱动。一、核心概念Trigger、Asset 与 AssetWatcher 如何协作在 Airflow 中一个监听外部队列的 DAG 由三要素组成Message Queue TriggerMessageQueueTrigger负责从外部消息队列如 AWS SQS、Kafka 或其他消息系统监听消息。它本身是一个统一封装——内部会根据你传入的scheme动态选择具体的 provider 触发器。Asset 与 AssetWatcherAsset抽象了外部实体例如那条 SQS 队列AssetWatcher则把一个具名 trigger挂到 asset 上。这个 name 用于标识哪个 trigger 对应哪个 asset是排查问题时的关键索引。事件驱动 DAGDAG 不再按固定周期运行而是当 asset 收到更新即队列里出现了新消息时执行。这种模式把调度权从时钟交给了消息适合下游数据处理、告警响应等对时效敏感、又不想高频空轮询的场景。二、等待队列消息SQS 完整示例官方文档给出的最小配置方式使用MessageQueueTrigger监听队列必填参数是scheme队列方案例如kafka、redispubsub、sqs。不同队列 provider 还支持各自的附加参数同时需要预先配置好连接使用相应的默认连接 ID——例如连接 AWS SQS 时连接 ID 应为aws_default。下面是系统测试中使用的真实示例 DAG位于 example_message_queue_trigger.pyfrom airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger from airflow.providers.standard.operators.empty import EmptyOperator from airflow.sdk import DAG, Asset, AssetWatcher # Define a trigger that listens to an external message queue (AWS SQS in this case) trigger MessageQueueTrigger(schemesqs, sqs_queuehttps://sqs.us-east-1.amazonaws.com/0123456789/Test) # Define an asset that watches for messages on the queue asset Asset(sqs_queue_asset, watchers[AssetWatcher(namesqs_watcher, triggertrigger)]) with DAG(dag_idexample_msgq_watcher, schedule[asset]) as dag: EmptyOperator(task_idtask)要点拆解schemesqs声明队列方案。sqs由 Amazon Provider 注册的SqsMessageQueueProvider认领其trigger_kwargs会把sqs_queue等参数透传给底层的SqsSensorTrigger见 sqs.py。sqs_queueSQS 队列的完整 URL作为 keyword argument 直接传给底层 trigger。Asset(..., watchers[AssetWatcher(name..., trigger...)])watcher 名称如sqs_watcher用于标识 trigger 与 asset 的对应关系。schedule[asset]把 asset 作为调度条件DAG 在该 asset 收到事件时触发运行。三、在 DAG 任务中使用消息 payload这是文档中最具实战价值的一段当消息到达时trigger 会把消息 payload 作为 trigger event 的一部分传递出来你可以在任务里通过triggering_asset_events参数直接读取from airflow.decorators import task from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger from airflow.sdk import DAG, Asset, AssetWatcher, chain # Define the asset with trigger trigger MessageQueueTrigger(schemekafka, topics[my-kafka-topic]) asset Asset(kafka_queue_asset, watchers[AssetWatcher(namekafka_watcher, triggertrigger)]) with DAG(dag_idexample_msgq_payload, schedule[asset]) as dag: task def process_message(triggering_asset_events): for event in triggering_asset_events[asset]: # Access the message payload payload event.extra[payload] # Process the payload as needed print(fReceived message: {payload}) chain(process_message())说明triggering_asset_events参数包含触发了本次 DAG 运行的事件按 asset 索引。每个 event 都带有extra字典消息 payload 存放在payload键下。例中使用schemekafka与topics[my-kafka-topic]Kafka 侧由KafkaMessageQueueProvider注册见 kafka/provider.yaml 中queues字段声明。这意味着下游任务可以直接拿到消息内容做业务处理而不是仅知道队列有新消息了。四、参数详解scheme、queue已废弃与 trigger_queue结合源码 msg_queue.pyMessageQueueTrigger.__init__接受以下参数参数必填说明scheme二选一队列方案如kafka、redispubsub、sqs用于 provider 匹配是推荐用法queue二选一已废弃。队列标识符URI 格式。若同时提供优先级高于scheme使用时会抛出AirflowProviderDeprecationWarningtrigger_queue否把该 trigger 分配到某个 triggerer 命名队列对应配置triggerer.queues_enabled与airflow triggerer --queues选项**kwargs否透传给底层 provider trigger 的参数如sqs_queue、topics、aws_conn_id等约束与行为queue与scheme必须提供其一否则抛出ValueError(Eitherqueueorschemeparameter must be provided.)。传入queue会触发废弃警告官方建议改用scheme并把配置以 keyword arguments 形式传入。trigger_queue与废弃的queue参数刻意命名区分因为它用于 triggerer 路由而非 broker 队列 URI重命名可避免破坏既有调用。trigger_queue的实际价值当不同队列 trigger 的资源消耗差异很大时可以用airflow triggerer --queues把重的监听任务隔离到独立队列避免相互饿死。五、源码剖析统一触发器如何路由到具体 providerMessageQueueTrigger的核心在trigger这个cached_propertymsg_queue.py其工作流程为动态发现模块加载时通过ProvidersManager().initialize_providers_queues()读取所有已安装 provider 的provider.yaml中声明的队列类实例化后放入MESSAGE_QUEUE_PROVIDERS。因此新 provider 只需在自己的provider.yaml中注册队列类即可被自动发现无需改动本模块。匹配若提供了queue旧方式逐个调用 provider 的queue_matches(queue_uri)做 URI 匹配如 SQS 的正则^https://sqs\.[^.]\.amazonaws\.com(\.cn)?/[0-9]/.定义于 sqs.py若提供scheme则调用scheme_matches(scheme)做精确字符串比较基类实现见 base_provider.py。冲突检测没有任何 provider 匹配时报ValueError并列出所有已注册 provider 名称多个 provider 同时匹配则报colliding错误。这种设计确保一个 scheme/queue 的归属唯一。构造底层 trigger匹配成功后取provider.trigger_class()生成真正的监听器实例如SqsSensorTrigger并把 kwargs 透传。事件透传run()方法只是async for event in self.trigger.run(): yield event即MessageQueueTrigger完全是个装饰器事件含extra[payload]原样从底层 trigger 流向 Asset 事件系统最终成为triggering_asset_events中的事件对象。从源码结构看serialize()直接委托给底层 trigger 的序列化因此 triggerer 重启恢复时行为与直接使用具体 trigger 完全一致。六、当前仓库中可用的队列 Provider 一览各 provider 在自己的provider.yaml的queues字段中注册队列类MessageQueueTrigger会自动发现它们。本仓库中当前注册的有Provider队列类schemeAmazonairflow.providers.amazon.aws.queues.sqs.SqsMessageQueueProvidersqsApache Kafkaairflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProviderkafkaGoogleairflow.providers.google.event_scheduling.events.pubsub.PubSubMessageQueueEventTriggerContainerpub/sub 相关IBM MQairflow.providers.ibm.mq.queues.mq.IBMMQMessageQueueProvideribm mq 相关Microsoft Azureairflow.providers.microsoft.azure.queues.asb.AzureServiceBusMessageBusMessageQueueProviderService Busazure service bus 相关Redisairflow.providers.redis.queues.redis.RedisPubSubMessageQueueProviderredispubsub具体注册行可见 amazon/provider.yaml、kafka/provider.yaml、redis/provider.yaml 等文件中的queues:段。可用的 scheme 集合取决于你安装了哪些 provider 包。七、运行前提与注意事项必须安装对应 provider 包MessageQueueTrigger本身不含任何队列实现若MESSAGE_QUEUE_PROVIDERS为空未安装任何队列类 providertrigger属性会抛出ValueError(No message queue providers are available.)并给出安装提示。连接配置各底层 trigger 依赖默认连接SQS 场景为aws_default。请先在 Airflow Connections 中配好凭据。需要 Triggererasset watcher 的 trigger 运行在 Triggerer 进程中事件驱动 DAG 要真正生效需确保 triggerer 已启动生产部署中如 Helm chart 等场景通常已包含该组件。废弃参数迁移如果你的旧代码使用了queuehttps://sqs....形式建议尽快迁移到schemesqs, sqs_queue...写法——源码表明queue优先级虽更高但已被标记废弃并将在未来版本移除。匹配唯一性不同 provider 的scheme_matches/queue_matches必须互不重叠一旦冲突会在创建 trigger 时立即报错这为多 provider 共存提供了保护。小结MessageQueueTrigger用一个 trigger、一个 scheme 参数抽象掉了 SQS、Kafka、Redis pub/sub 等多种消息系统的差异配合AssetAssetWatcherschedule[asset]的组合你可以在不写任何 sensor 长轮询代码的情况下构建出消息到达即触发的 DAG并通过triggering_asset_events[asset]中事件的extra[payload]直接消费消息内容。相关实现集中在 providers/common/messaging 目录单元与系统测试分别在 test_msg_queue.py 和 example_message_queue_trigger.py可作为行为参考。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
