Apache Airflow 集成 Amazon SQSSqsPublishOperator 与 SqsSensor 消息收发实战指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAmazon Simple Queue Service (SQS) 是 AWS 提供的全托管消息队列服务用于在微服务、分布式系统与 Serverless 应用之间解耦和伸缩消息传递。本文以 Apache Airflow 官方 Amazon Provider本仓库providers/amazon目录为依托围绕 SQS 使用指南 的完整脉络深入讲解如何用SqsPublishOperator向队列发布消息、用SqsSensor读取并消费消息并结合源码剖析参数背后的真实调用链让你既能照抄可运行的 DAG 示例也能理解每一个参数在底层如何影响 boto3 调用。为什么在 Airflow 中使用 Amazon SQS在 Airflow 工作流中SQS 最常见的应用场景是充当任务之间的消息总线上游任务把消息写入队列下游任务或独立的消费进程从队列拉取消息从而实现组件解耦、削峰填谷以及跨系统异步通信。SQS 作为全托管服务免去了自建消息中间件的运维负担可以按任意规模发送、存储和接收消息且不要求消费方始终在线——消息会暂存于队列中直到被消费。在 sqs.rst 中官方提供了一条完整的发布 → 读取链路使用SqsPublishOperator向队列发布消息使用SqsSensor阻塞等待并读取消息两者配合即可在 Airflow 内实现可靠的队列驱动型任务编排。前置准备安装与连接配置在编写任何 SQS 相关任务前需要完成三项准备见 prerequisite_tasks.rst在 AWS 控制台或通过 AWS CLI 创建必要的资源如 SQS 队列安装 Amazon Providerpip install apache-airflow[amazon]配置 AWS 连接详见 AWS 连接指南核心是设置aws_conn_id默认值aws_default以及可选的region_name等连接参数。除连接外所有 AWS 算子/传感器还共享一组通用参数见 generic_parameters.rst包括参数说明默认值aws_conn_id引用 AWS 连接的 ID设为None时直接采用 boto3 默认凭证行为aws_defaultregion_nameAWS 区域名省略时回退到连接中的region_nameNoneverify是否校验 SSL 证书可为False或 CA 证书 bundle 路径Nonebotocore_config构造botocore.config.Config的字典可配置重试、超时等None其中botocore_config可以这样使用用于避免节流异常并配置超时{ signature_version: unsigned, s3: { us_east_1_regional_endpoint: True, }, retries: { mode: standard, max_attempts: 10, }, connect_timeout: 300, read_timeout: 300, tcp_keepalive: True, }需要注意若显式传入空字典{}会覆盖连接中配置的botocore_config。发布消息SqsPublishOperator 完整解析官方示例向队列发布消息官方文档 指出向 SQS 队列发布消息使用SqsPublishOperator。以 example_sqs.py 中的真实系统测试为例发布任务把任务实例信息写入名为Airflow-Example-Queue风格实际按环境变量生成的{env_id}-example-queue的队列publish_to_queue_1 SqsPublishOperator( task_idpublish_to_queue_1, sqs_queuesqs_queue, message_content{{ task_instance }}, ) publish_to_queue_2 SqsPublishOperator( task_idpublish_to_queue_2, sqs_queuesqs_queue, message_content{{ task_instance }}, )参数与底层实现SqsPublishOperator定义于 operators/sqs.py其核心参数如下参数说明默认值是否模板化sqs_queueSQS 队列 URL必填—✅message_content消息正文必填—✅message_attributes消息附加属性对应botocore SQS.send_message的 attributes 参数None✅delay_seconds消息延迟秒数对应DelaySeconds0✅message_group_id仅适用于 FIFO 队列对应MessageGroupIdNone✅message_deduplication_id仅适用于 FIFO 队列对应MessageDeduplicationIdNone✅它继承自AwsBaseOperator并通过aws_hook_class SqsHook绑定底层 SQS 客户端。execute()方法把上述参数直接透传给SqsHook.send_message()见 hooks/sqs.py后者通过_build_msg_params构造 boto3send_message的请求参数并使用prune_dict剔除值为None的键——这意味着不传message_group_id时该参数不会出现在 API 请求中而是采用 boto3 默认行为。发送成功后任务日志会记录返回结果包含MessageId、MD5OfMessageBody等并把结果 dict 作为返回值推送到 XCom。模板化能力template_fields声明了sqs_queue、message_content、delay_seconds、message_attributes、message_group_id、message_deduplication_id六个可渲染字段这意味着你可以用 Jinja 模板动态生成消息内容如示例中的{{ task_instance }}。message_attributes还配置了template_fields_renderers {message_attributes: json}在 Airflow UI 的 Rendered 视图里会以 JSON 格式展示。关于 FIFO 队列的注意点单元测试 operators/test_sqs.py 覆盖了 FIFO 场景向 FIFO 队列发送消息时必须提供message_group_id否则 boto3 会抛错test_execute_failure_fifo_queue与test_execute_success_fifo_queue分别验证了缺失与提供该参数时的行为test_deduplication_failure则验证了重复使用同一message_deduplication_id会触发去重失败。因此当你的目标队列是.fifo结尾的 FIFO 队列时务必同时指定这两个参数。读取消息SqsSensor 完整解析官方示例读取并消费消息读取消息使用SqsSensor它会持续轮询直到队列中的消息被读取耗尽才返回成功。官方示例同样来自 example_sqs.pyread_from_queue SqsSensor( task_idread_from_queue, sqs_queuesqs_queue, ) # Retrieve multiple batches of messages from SQS. # The SQS API only returns a maximum of 10 messages per poll. read_from_queue_in_batch SqsSensor( task_idread_from_queue_in_batch, sqs_queuesqs_queue, # Get maximum 10 messages each poll max_messages10, # Combine 3 polls before returning results num_batches3, )参数与消费语义SqsSensor定义于 sensors/sqs.py其核心参数包括参数说明默认值sqs_queueSQS 队列 URL必填可模板化—max_messages每次轮询SQS API 调用最多拉取的消息数5num_batches每次 poke 内调用 SQS API 的次数用于串行聚合多批消息1wait_time_seconds单次 receive 的等待秒数对应WaitTimeSeconds支持长轮询1visibility_timeout可见性超时期间 SQS 阻止其他消费者接收/处理该消息Nonemessage_filtering消息过滤方式None、literal、jsonpath、jsonpath-extNonemessage_filtering_match_values过滤匹配的目标值集合Nonemessage_filtering_config过滤附加配置如 JSONPath 表达式Nonedelete_message_on_reception消费后立即从队列删除消息False时消息保留需手动清理Truedeferrable是否以可延迟Deferrable模式运行需要安装aiobotocore见下文其中deferrable的默认值取自 Airflow 配置项operators.default_deferrableconf.getboolean(operators, default_deferrable, fallbackFalse)即默认False但可以在 Airflow 配置文件中全局打开。消费语义从poke()源码可见按num_batches循环调用poll_sqs()封装receive_message最多返回 10 条/次受max_messages限制对每批结果执行process_response()过滤见下节若delete_message_on_receptionTrue则用delete_message_batch批量删除已接收消息——删除失败会抛出AirflowException只要累计收到消息就通过 XCom 以messages为 key 推送消息列表并返回True否则返回False继续等待下次 poke。这一读即删的默认语义保证消息最多被消费一次是传感器能读到队列耗尽的关键。若设置delete_message_on_receptionFalse消息会保留在队列中单元测试 sensors/test_sqs.py 的test_poke_do_not_delete_message_on_received专门验证了此时不会调用delete_message_batch适合看一下队列里有什么的场景但需自行负责清理。消息过滤literal / jsonpath / jsonpath-ext过滤逻辑实现在 utils/sqs.py 的filter_messages中literal消息Body必须完全等于message_filtering_match_values中的某个值filter_messages_literal。源码中若指定了literal却未提供匹配值SqsSensor.__init__会直接抛出TypeError。jsonpath将消息Body反序列化为 JSON用jsonpath_ng.parse(message_filtering_config)解析表达式匹配表达式结果命中message_filtering_match_values的消息filter_messages_jsonpath。例如配置foo[*].baz。jsonpath-ext与jsonpath相同但使用jsonpath_ng.ext.parse支持更扩展的查询语法如$.key $.value拼接字段。相关单元测试test_poke_message_filtering_literal_values、test_poke_message_filtering_jsonpath、test_poke_message_filtering_jsonpath_ext等覆盖了每种过滤模式验证了过滤后仅删除匹配消息的行为。轮询节奏与长轮询wait_time_seconds直接映射到 boto3receive_message的WaitTimeSeconds参数。将其调大SQS 长轮询上限 20 秒可以减少空轮询次数、降低 API 调用成本visibility_timeout仅在显式指定时才加入请求参数源码中if self.visibility_timeout is not None用于控制消息在被消费期间的锁定期。可延迟模式deferrable 与 SqsSensorTriggerSqsSensor设置deferrableTrue后会从阻塞式轮询切换为异步可延迟模式任务在execute()中通过self.defer(...)挂起释放 Worker 资源交给SqsSensorTrigger异步处理仅在收到消息或超时才唤醒任务。触发器实现在 triggers/sqs.py使用SqsHook.get_async_conn()建立异步连接因此需要安装aiobotocorerun()中循环执行异步poke()调用receive_message拉取消息、按num_batches聚合、默认delete_message_batch删除拿到消息就yield TriggerEvent({status: success, message_batch: result})否则await asyncio.sleep(waiter_delay)后重试waiter_delay默认 60 秒由传感器poke_interval传入唤醒后execute_complete()校验事件状态并把message_batch通过 XCom 的messageskey 推送。这种模式适合队列长时间为空、无需一直占用 Worker 的场景可显著节省调度资源。单元测试test_sqs_deferrable与test_fail_execute_complete验证了 deferrable 路径的成功与失败分支。完整实战一条可运行的 SQS 消费 DAG结合 example_sqs.py 的完整结构一个可运行的 SQS 发布-消费 DAG 通常包含创建队列测试上下文→ 发布 → 读取 → 发布 → 批量读取 → 删除队列清理。核心骨架如下from airflow.providers.amazon.aws.operators.sqs import SqsPublishOperator from airflow.providers.amazon.aws.sensors.sqs import SqsSensor from airflow.providers.amazon.aws.hooks.sqs import SqsHook # 创建队列返回 QueueUrl sqs_queue SqsHook().create_queue(queue_namemy-example-queue)[QueueUrl] # 发布两条消息 publish_1 SqsPublishOperator( task_idpublish_1, sqs_queuesqs_queue, message_content{{ task_instance }}, ) # 读取消息读即删读到耗尽为止 read_all SqsSensor(task_idread_all, sqs_queuesqs_queue) # 批量读取每次最多 10 条聚合 3 次轮询 read_batch SqsSensor( task_idread_batch, sqs_queuesqs_queue, max_messages10, num_batches3, )运行约束与提示该示例作为系统测试运行airflow tasks test example_sqs ...或通过 pytest 运行create_queue/delete_queue承担了队列的创建与清理避免手工准备资源也可以直接在 AWS 控制台或 CLI 中预先建好队列再把sqs_queue换成固定 URL。系统测试依赖AIRFLOW_V_3_0_PLUS做版本兼容分支说明该 Provider 同时兼容 Airflow 2.10 与 3.x本仓库核心为 Airflow 3 时代代码。发布后的返回值含MessageId会进入 XCom可被下游任务通过ti.xcom_pull消费传感器读取到的消息列表存放在messageskey 下。小结本文基于 sqs.rst 完整梳理了 Airflow 与 Amazon SQS 集成的两条主线SqsPublishOperator负责发布支持模板化、FIFO 参数、消息属性SqsSensor负责读取消费支持批量轮询、消息过滤、可见性超时、读即删以及可延迟模式。对应的源码入口分别为 operators/sqs.py、sensors/sqs.py、hooks/sqs.py、triggers/sqs.py 与 utils/sqs.py单元测试见 operators/test_sqs.py 与 sensors/test_sqs.py。掌握这些细节后你就可以在真实工作流中放心地以 SQS 作为任务间消息总线实现可靠的异步解耦与削峰。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Airflow 集成 Amazon SQS：SqsPublishOperator 与 SqsSensor 消息收发实战指南
Apache Airflow 集成 Amazon SQSSqsPublishOperator 与 SqsSensor 消息收发实战指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAmazon Simple Queue Service (SQS) 是 AWS 提供的全托管消息队列服务用于在微服务、分布式系统与 Serverless 应用之间解耦和伸缩消息传递。本文以 Apache Airflow 官方 Amazon Provider本仓库providers/amazon目录为依托围绕 SQS 使用指南 的完整脉络深入讲解如何用SqsPublishOperator向队列发布消息、用SqsSensor读取并消费消息并结合源码剖析参数背后的真实调用链让你既能照抄可运行的 DAG 示例也能理解每一个参数在底层如何影响 boto3 调用。为什么在 Airflow 中使用 Amazon SQS在 Airflow 工作流中SQS 最常见的应用场景是充当任务之间的消息总线上游任务把消息写入队列下游任务或独立的消费进程从队列拉取消息从而实现组件解耦、削峰填谷以及跨系统异步通信。SQS 作为全托管服务免去了自建消息中间件的运维负担可以按任意规模发送、存储和接收消息且不要求消费方始终在线——消息会暂存于队列中直到被消费。在 sqs.rst 中官方提供了一条完整的发布 → 读取链路使用SqsPublishOperator向队列发布消息使用SqsSensor阻塞等待并读取消息两者配合即可在 Airflow 内实现可靠的队列驱动型任务编排。前置准备安装与连接配置在编写任何 SQS 相关任务前需要完成三项准备见 prerequisite_tasks.rst在 AWS 控制台或通过 AWS CLI 创建必要的资源如 SQS 队列安装 Amazon Providerpip install apache-airflow[amazon]配置 AWS 连接详见 AWS 连接指南核心是设置aws_conn_id默认值aws_default以及可选的region_name等连接参数。除连接外所有 AWS 算子/传感器还共享一组通用参数见 generic_parameters.rst包括参数说明默认值aws_conn_id引用 AWS 连接的 ID设为None时直接采用 boto3 默认凭证行为aws_defaultregion_nameAWS 区域名省略时回退到连接中的region_nameNoneverify是否校验 SSL 证书可为False或 CA 证书 bundle 路径Nonebotocore_config构造botocore.config.Config的字典可配置重试、超时等None其中botocore_config可以这样使用用于避免节流异常并配置超时{ signature_version: unsigned, s3: { us_east_1_regional_endpoint: True, }, retries: { mode: standard, max_attempts: 10, }, connect_timeout: 300, read_timeout: 300, tcp_keepalive: True, }需要注意若显式传入空字典{}会覆盖连接中配置的botocore_config。发布消息SqsPublishOperator 完整解析官方示例向队列发布消息官方文档 指出向 SQS 队列发布消息使用SqsPublishOperator。以 example_sqs.py 中的真实系统测试为例发布任务把任务实例信息写入名为Airflow-Example-Queue风格实际按环境变量生成的{env_id}-example-queue的队列publish_to_queue_1 SqsPublishOperator( task_idpublish_to_queue_1, sqs_queuesqs_queue, message_content{{ task_instance }}, ) publish_to_queue_2 SqsPublishOperator( task_idpublish_to_queue_2, sqs_queuesqs_queue, message_content{{ task_instance }}, )参数与底层实现SqsPublishOperator定义于 operators/sqs.py其核心参数如下参数说明默认值是否模板化sqs_queueSQS 队列 URL必填—✅message_content消息正文必填—✅message_attributes消息附加属性对应botocore SQS.send_message的 attributes 参数None✅delay_seconds消息延迟秒数对应DelaySeconds0✅message_group_id仅适用于 FIFO 队列对应MessageGroupIdNone✅message_deduplication_id仅适用于 FIFO 队列对应MessageDeduplicationIdNone✅它继承自AwsBaseOperator并通过aws_hook_class SqsHook绑定底层 SQS 客户端。execute()方法把上述参数直接透传给SqsHook.send_message()见 hooks/sqs.py后者通过_build_msg_params构造 boto3send_message的请求参数并使用prune_dict剔除值为None的键——这意味着不传message_group_id时该参数不会出现在 API 请求中而是采用 boto3 默认行为。发送成功后任务日志会记录返回结果包含MessageId、MD5OfMessageBody等并把结果 dict 作为返回值推送到 XCom。模板化能力template_fields声明了sqs_queue、message_content、delay_seconds、message_attributes、message_group_id、message_deduplication_id六个可渲染字段这意味着你可以用 Jinja 模板动态生成消息内容如示例中的{{ task_instance }}。message_attributes还配置了template_fields_renderers {message_attributes: json}在 Airflow UI 的 Rendered 视图里会以 JSON 格式展示。关于 FIFO 队列的注意点单元测试 operators/test_sqs.py 覆盖了 FIFO 场景向 FIFO 队列发送消息时必须提供message_group_id否则 boto3 会抛错test_execute_failure_fifo_queue与test_execute_success_fifo_queue分别验证了缺失与提供该参数时的行为test_deduplication_failure则验证了重复使用同一message_deduplication_id会触发去重失败。因此当你的目标队列是.fifo结尾的 FIFO 队列时务必同时指定这两个参数。读取消息SqsSensor 完整解析官方示例读取并消费消息读取消息使用SqsSensor它会持续轮询直到队列中的消息被读取耗尽才返回成功。官方示例同样来自 example_sqs.pyread_from_queue SqsSensor( task_idread_from_queue, sqs_queuesqs_queue, ) # Retrieve multiple batches of messages from SQS. # The SQS API only returns a maximum of 10 messages per poll. read_from_queue_in_batch SqsSensor( task_idread_from_queue_in_batch, sqs_queuesqs_queue, # Get maximum 10 messages each poll max_messages10, # Combine 3 polls before returning results num_batches3, )参数与消费语义SqsSensor定义于 sensors/sqs.py其核心参数包括参数说明默认值sqs_queueSQS 队列 URL必填可模板化—max_messages每次轮询SQS API 调用最多拉取的消息数5num_batches每次 poke 内调用 SQS API 的次数用于串行聚合多批消息1wait_time_seconds单次 receive 的等待秒数对应WaitTimeSeconds支持长轮询1visibility_timeout可见性超时期间 SQS 阻止其他消费者接收/处理该消息Nonemessage_filtering消息过滤方式None、literal、jsonpath、jsonpath-extNonemessage_filtering_match_values过滤匹配的目标值集合Nonemessage_filtering_config过滤附加配置如 JSONPath 表达式Nonedelete_message_on_reception消费后立即从队列删除消息False时消息保留需手动清理Truedeferrable是否以可延迟Deferrable模式运行需要安装aiobotocore见下文其中deferrable的默认值取自 Airflow 配置项operators.default_deferrableconf.getboolean(operators, default_deferrable, fallbackFalse)即默认False但可以在 Airflow 配置文件中全局打开。消费语义从poke()源码可见按num_batches循环调用poll_sqs()封装receive_message最多返回 10 条/次受max_messages限制对每批结果执行process_response()过滤见下节若delete_message_on_receptionTrue则用delete_message_batch批量删除已接收消息——删除失败会抛出AirflowException只要累计收到消息就通过 XCom 以messages为 key 推送消息列表并返回True否则返回False继续等待下次 poke。这一读即删的默认语义保证消息最多被消费一次是传感器能读到队列耗尽的关键。若设置delete_message_on_receptionFalse消息会保留在队列中单元测试 sensors/test_sqs.py 的test_poke_do_not_delete_message_on_received专门验证了此时不会调用delete_message_batch适合看一下队列里有什么的场景但需自行负责清理。消息过滤literal / jsonpath / jsonpath-ext过滤逻辑实现在 utils/sqs.py 的filter_messages中literal消息Body必须完全等于message_filtering_match_values中的某个值filter_messages_literal。源码中若指定了literal却未提供匹配值SqsSensor.__init__会直接抛出TypeError。jsonpath将消息Body反序列化为 JSON用jsonpath_ng.parse(message_filtering_config)解析表达式匹配表达式结果命中message_filtering_match_values的消息filter_messages_jsonpath。例如配置foo[*].baz。jsonpath-ext与jsonpath相同但使用jsonpath_ng.ext.parse支持更扩展的查询语法如$.key $.value拼接字段。相关单元测试test_poke_message_filtering_literal_values、test_poke_message_filtering_jsonpath、test_poke_message_filtering_jsonpath_ext等覆盖了每种过滤模式验证了过滤后仅删除匹配消息的行为。轮询节奏与长轮询wait_time_seconds直接映射到 boto3receive_message的WaitTimeSeconds参数。将其调大SQS 长轮询上限 20 秒可以减少空轮询次数、降低 API 调用成本visibility_timeout仅在显式指定时才加入请求参数源码中if self.visibility_timeout is not None用于控制消息在被消费期间的锁定期。可延迟模式deferrable 与 SqsSensorTriggerSqsSensor设置deferrableTrue后会从阻塞式轮询切换为异步可延迟模式任务在execute()中通过self.defer(...)挂起释放 Worker 资源交给SqsSensorTrigger异步处理仅在收到消息或超时才唤醒任务。触发器实现在 triggers/sqs.py使用SqsHook.get_async_conn()建立异步连接因此需要安装aiobotocorerun()中循环执行异步poke()调用receive_message拉取消息、按num_batches聚合、默认delete_message_batch删除拿到消息就yield TriggerEvent({status: success, message_batch: result})否则await asyncio.sleep(waiter_delay)后重试waiter_delay默认 60 秒由传感器poke_interval传入唤醒后execute_complete()校验事件状态并把message_batch通过 XCom 的messageskey 推送。这种模式适合队列长时间为空、无需一直占用 Worker 的场景可显著节省调度资源。单元测试test_sqs_deferrable与test_fail_execute_complete验证了 deferrable 路径的成功与失败分支。完整实战一条可运行的 SQS 消费 DAG结合 example_sqs.py 的完整结构一个可运行的 SQS 发布-消费 DAG 通常包含创建队列测试上下文→ 发布 → 读取 → 发布 → 批量读取 → 删除队列清理。核心骨架如下from airflow.providers.amazon.aws.operators.sqs import SqsPublishOperator from airflow.providers.amazon.aws.sensors.sqs import SqsSensor from airflow.providers.amazon.aws.hooks.sqs import SqsHook # 创建队列返回 QueueUrl sqs_queue SqsHook().create_queue(queue_namemy-example-queue)[QueueUrl] # 发布两条消息 publish_1 SqsPublishOperator( task_idpublish_1, sqs_queuesqs_queue, message_content{{ task_instance }}, ) # 读取消息读即删读到耗尽为止 read_all SqsSensor(task_idread_all, sqs_queuesqs_queue) # 批量读取每次最多 10 条聚合 3 次轮询 read_batch SqsSensor( task_idread_batch, sqs_queuesqs_queue, max_messages10, num_batches3, )运行约束与提示该示例作为系统测试运行airflow tasks test example_sqs ...或通过 pytest 运行create_queue/delete_queue承担了队列的创建与清理避免手工准备资源也可以直接在 AWS 控制台或 CLI 中预先建好队列再把sqs_queue换成固定 URL。系统测试依赖AIRFLOW_V_3_0_PLUS做版本兼容分支说明该 Provider 同时兼容 Airflow 2.10 与 3.x本仓库核心为 Airflow 3 时代代码。发布后的返回值含MessageId会进入 XCom可被下游任务通过ti.xcom_pull消费传感器读取到的消息列表存放在messageskey 下。小结本文基于 sqs.rst 完整梳理了 Airflow 与 Amazon SQS 集成的两条主线SqsPublishOperator负责发布支持模板化、FIFO 参数、消息属性SqsSensor负责读取消费支持批量轮询、消息过滤、可见性超时、读即删以及可延迟模式。对应的源码入口分别为 operators/sqs.py、sensors/sqs.py、hooks/sqs.py、triggers/sqs.py 与 utils/sqs.py单元测试见 operators/test_sqs.py 与 sensors/test_sqs.py。掌握这些细节后你就可以在真实工作流中放心地以 SQS 作为任务间消息总线实现可靠的异步解耦与削峰。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考