Apache Airflow Kafka Provider 包深度指南安装依赖、连接配置与 Hook/Operator/Sensor/Trigger 源码级解析【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本文以apache-airflow-providers-apache-kafka2.0.0 版本文档为核心系统讲解该 Provider 包的版本要求、依赖矩阵与安装方式并结合仓库源码深入解析 Kafka 连接的安全配置机制回调白名单、MSK IAM 自动认证、核心组件Hook、Operator、Sensor、Trigger、消息队列、事件插件的实现细节读完后可独立完成 Kafka Provider 的部署、连接配置与 DAG 实战开发。一、包概览apache.kafka Provider 是什么apache-airflow-providers-apache-kafka是 Apache Airflow 官方的 Apache Kafka Provider 发行包当前版本2.0.0负责让 Airflow 工作流与 Kafka 集群进行生产、消费、监听与事件驱动交互。该包的所有类都位于airflow.providers.apache.kafkaPython 包中通过标准 provider 机制注册到 Airflow。从 provider 元数据定义文件 可以确认这个包对外暴露了哪些组件类型组件类型模块Operatorsairflow.providers.apache.kafka.operators.consume、airflow.providers.apache.kafka.operators.produceHookshooks.base、hooks.client、hooks.consume、hooks.produceSensorairflow.providers.apache.kafka.sensors.kafkaTriggerstriggers.await_message、triggers.msg_queue消息队列 Providerairflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider插件kafka_event_producerKafkaEventProducerPluginAsset URI schemekafka://含 sanitize、create_asset 与 OpenLineage 转换器连接类型kafka默认 Hook 为KafkaBaseHook其中kafka://方案注册意味着 Airflow 的 Asset数据集机制可以直接用kafka://开头的 URI 表示一个 Kafka 主题并在 OpenLineage 数据血缘中转换。这些注册信息最终通过 pyproject.toml 中的 entry pointairflow.providers.apache.kafka.get_provider_info:get_provider_info被 Airflow 在启动时发现。官方文档的入口是 Kafka Provider 文档首页其下包含连接、Hook、Operator、消息队列、Sensor、Trigger 等使用指南以及 配置参考 与 Python API 参考。二、安装与版本要求2.1 安装方式在已有 Airflow 安装之上直接通过 pip 安装即可pip install apache-airflow-providers-apache-kafka如需源码安装可参考 从源码安装 Provider 的文档。2.2 最低版本与依赖矩阵该 Provider 发行包支持的最低 Apache Airflow 版本为2.11.0。完整依赖要求与 pyproject.toml 中dependencies一致PIP 包版本要求apache-airflow2.11.0apache-airflow-providers-common-compat1.12.0asgiref2.3.0Python 3.143.11.1Python 3.14confluent-kafka2.6.0Python 3.142.13.2Python 3.14可以看出该包要求 Python 3.10底层客户端库为confluent-kafkalibrdkafka 的 Python 绑定并且针对 Python 3.14 单独提高了asgiref与confluent-kafka的版本下限。2.3 跨 Provider 依赖Cross-provider dependencies以下依赖用于启用包的完整功能需要从 PyPI 安装对应的 provider 发行包可通过 extras 一次性装好pip install apache-airflow-providers-apache-kafka[common.messaging]依赖包Extra 名称apache-airflow-providers-common-messagingcommon.messagingapache-airflow-providers-googlegoogle2.4 可选依赖Optional extras这些 extras 安装可选的第三方库以启用额外功能从 PyPI 安装时使用pip install apache-airflow-providers-apache-kafka[google]Extra引入的依赖用途googleapache-airflow-providers-googleGoogle Managed Kafka 的自动 OAuth token 认证mskaws-msk-iam-sasl-signer-python1.0.1Amazon MSK IAM 自动认证common.messagingapache-airflow-providers-common-messaging2.0.0通用消息队列框架MessageQueueTrigger等官方发布的 sdist / wheel 包可在 Apache 官方下载站点获取并校验 sha512 与 asc 签名包名与版本号以当前发行版2.0.0为准。三、Kafka 连接配置Config Dict、回调白名单与托管认证3.1 连接类型与默认连接 IDKafka 连接类型基于confluent-kafka库extra字段是一个 JSON 可序列化的配置字典即 librdkafka 的全部配置项如bootstrap.servers、group.id、enable.auto.commit等。在 Airflow UI 中创建连接时host、port、schema、login、password字段会被隐藏extra字段会被重命名为Config Dict——这一 UI 行为正是由 KafkaBaseHook 的get_ui_field_behaviour方法返回的hidden_fields、relabeling与placeholders占位提示{bootstrap.servers: localhost:9092, group.id: my-group}驱动的。Kafka 连接配置界面所有 Kafka Hook 与 Operator 默认使用连接 IDkafka_default。这个默认连接极其简略只适合最基础的测试生产环境应创建自己的连接。最小有效配置的校验逻辑在源码中很明确bootstrap.servers键必须存在且有值否则直接抛出ValueError见 base.py 的_build_config。一个典型的连接extra示例也见于 系统测试示例 DAG 中通过AIRFLOW_CONN_*环境变量注入的连接{ bootstrap.servers: broker:9092, group.id: my-group, enable.auto.commit: false, auto.offset.reset: beginning }详细文档见 Kafka 连接指南。3.2 回调白名单字符串回调的安全边界连接extra中error_cb、throttle_cb、stats_cb、log_cb、oauth_cb、on_commit这六项 librdkafka 配置允许以点号路径字符串如module.callback_func形式提供Hook 会在构建客户端前将其解析为可调用对象。可解析的键名常量定义在 base.pyCALLBACK_CONFIG_KEYS (error_cb, throttle_cb, stats_cb, log_cb, oauth_cb, on_commit)从源码的_resolve_callbacks实现base.py L106-L145可以确认其安全策略回调只在完整导入路径模块 属性如my_company.kafka.auth.oauth_cb被列入[apache_kafka] callback_allowlist配置时才被导入执行条目必须与连接值精确匹配只写裸模块名如my_company.kafka.auth不会授权该模块内的任何可调用对象。白名单为空默认值时任何字符串形式的回调都会被拒绝并抛出ValueError同时记录 warning 日志。托管认证Amazon MSK IAM、Google Managed Kafka由 Hook 内部注入回调不经过白名单不受其影响。这解释了 连接文档中的警告该机制是为防止连接配置中的恶意回调被导入执行。使用自定义回调时需在 Airflow 配置中显式添加[apache_kafka] callback_allowlist my_company.kafka.auth.oauth_cb3.3 Amazon MSK IAM 自动认证对 Amazon MSK含 provisioned 与 serverless集群连接extra只需配置SASL_SSLOAUTHBEARER{ bootstrap.servers: boot-abcde1.c2.kafka-serverless.us-east-1.amazonaws.com:9098, security.protocol: SASL_SSL, sasl.mechanism: OAUTHBEARER, group.id: my-group }从 base.py 的_maybe_add_msk_iam_oauth可看到自动注入逻辑的完整判断链先用正则MSK_BOOTSTRAP_SERVERS_REGEXL43-L46识别 MSK 端点命名模式*.kafka.region.amazonaws.com与*.kafka-serverless.region.amazonaws.com含中国区.amazonaws.com.cn后缀并捕获 region仅当sasl.mechanism或sasl.mechanisms为OAUTHBEARER且匹配到 MSK 端点时才注入_msk_iam_oauth_cb(region)作为oauth_cb用户显式提供的oauth_cb永远优先不会被覆盖未安装mskextra 时抛出AirflowOptionalProviderFeatureException提示执行pip install apache-airflow-providers-apache-kafka[msk]AWS 凭证由 signer 通过标准 AWS 凭证链环境变量、共享配置文件、实例/任务 IAM 角色等解析。同理当bootstrap.servers指向 Google Managed Kafka包含cloud.goog与managedkafka字样时_build_config会自动导入 Google Provider 的ManagedKafkaHook并注入其get_confluent_token作为oauth_cbbase.py L159-L183需要预装googleProvider 14.1.0。值得注意的设计是test_connectionbase.py L230-L243调用同一个_build_config()来构建配置因此 UI 里“测试连接”与真实任务运行使用完全一致的回调解析与托管认证行为避免“UI 测试通过、任务却失败”的偏差。四、核心 HooksAdminClient、Producer、ConsumerHook 文档 列出了本包四个 HookHook模块说明KafkaBaseHook旧名KafkaHookhooks.base与 Kafka 交互的基类自定义 Kafka Hook 应继承它默认连接 ID 为kafka_defaultKafkaAdminClientHookhooks.client对 Kafka AdminClient 的封装用于集群管理操作KafkaConsumerHookhooks.consume创建 Kafka Consumer被ConsumeFromTopicOperator与AwaitMessageTrigger使用KafkaProducerHookhooks.produce创建 Kafka Producer被ProduceToTopicOperator使用从 base.py 可确认KafkaBaseHook的关键属性conn_type kafka、default_conn_name kafka_default、hook_name Apache Kafka其get_conn是一个cached_property返回以完整解析后配置构建的AdminClient。此外还有KafkaAuthenticationError自定义异常用于 Kafka 认证失败场景。五、OperatorsProduceToTopicOperator 与 ConsumeFromTopicOperatorOperator 文档 覆盖两个 Operator官方用法示例直接内嵌自 example_dag_hello_kafka.py5.1 ProduceToTopicOperator向主题生产消息t1 ProduceToTopicOperator( kafka_config_idt1-3, task_idproduce_to_topic, topictest_1, producer_functionexample_dag_hello_kafka.producer_function, )producer_function是用户提供的生成器/函数产出的 (key, value) 消息对会被发布到指定主题可以传点号字符串如上例DAG 序列化友好也可以直接传可调用对象。示例中对应的生成器def producer_function(): for i in range(20): yield (json.dumps(i), json.dumps(i 1))5.2 ConsumeFromTopicOperator批量消费并处理消息t2 ConsumeFromTopicOperator( kafka_config_idt2, task_idconsume_from_topic, topics[test_1], apply_functionexample_dag_hello_kafka.consumer_function, apply_function_kwargs{prefix: consumed:::}, commit_cadenceend_of_batch, max_messages10, max_batch_size2, )行为要点来自官方文档与示例 DAGOperator 创建一个 Kafka Consumer按批读取消息并用apply_function逐条处理或apply_function_batch整批处理直到到达日志末尾或达到max_messages上限提交语义若设置了commit_cadence参数必须确保连接配置中显式将enable.auto.commit设为false。默认enable.auto.commit为trueconsumer 每 5 秒自动提交 offset会覆盖commit_cadence定义的提交行为——这也是示例 DAG 中每个消费连接都显式带enable.auto.commit: False的原因设置return_apply_function_resultsTrue可返回apply_function每条消息返回的非None值列表按消费顺序返回值走普通 XCom 通道避免返回大负载该选项不适用于apply_function_batch。六、SensorsAwaitMessageSensor 与 AwaitMessageTriggerFunctionSensorSensor 文档 介绍两个可 defer 的 Sensor6.1 AwaitMessageSensor等待满足条件的消息t5 AwaitMessageSensor( kafka_config_idt5, task_idawaiting_message, topics[test_1], apply_functionexample_dag_hello_kafka.await_function, xcom_push_keyretrieved_message, )Sensor 会持续消费主题直到apply_function对某条消息返回真值此时触发TriggerEvent并成功完成。示例中的判断函数是“值能被 5 整除”def await_function(message): if json.loads(message.value()) % 5 0: return f Got the following message: {json.loads(message.value())}6.2 AwaitMessageTriggerFunctionSensor命中后触发回调再继续等待与上者不同该 Sensor 在消费到满足条件的消息后会调用event_triggered_function提供的可调用对象然后再次 defer 继续消费适合“事件监听器”类场景示例见 example_dag_event_listener.py。重要约束apply_function必须以点号字符串形式提供而不能传函数对象——因为 trigger 参数会被序列化进元数据库函数在triggerer 进程中被导入执行模块必须在那里可导入且修改函数需要重启 triggerer。详细要求见 消息队列文档中的 apply_function 一节。七、TriggersAwaitMessageTrigger 与 KafkaMessageQueueTriggerTrigger 文档 定义两个 triggerAwaitMessageTriggerairflow.providers.apache.kafka.triggers.await_message在 triggerer 中消费轮询到的 Kafka 消息并交给apply_function处理可调用返回任意数据时即 raiseTriggerEvent把任务唤醒。这是AwaitMessageSensor的运行时实现。KafkaMessageQueueTriggerairflow.providers.apache.kafka.triggers.msg_queueKafka 消息队列的专用接口类继承通用airflow.providers.common.messaging.triggers.msg_queue.MessageQueueTrigger与KafkaMessageQueueProvider配合提供 Kafka 消息队列操作的具体接口。八、消息队列集成KafkaMessageQueueProvider 与 MessageQueueTrigger本包还实现了 Kafka 消息队列 Providerairflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider是通用BaseMessageQueueProvider的 Kafka 后端允许在 Airflow 工作流中用 Kafka 主题作为消息队列并通过统一的MessageQueueTrigger接口收发。典型用法用 Kafka 主题上的新消息触发 DAGkafka://Asset 方案 监听器官方示例见 example_dag_kafka_message_queue_trigger.pyfrom airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger trigger MessageQueueTrigger( schemekafka, topics[my_topic], apply_functionmy_package.my_module.my_function, apply_function_args[received:], apply_function_kwargs{threshold: 100}, )apply_function的调用约定对触发器类组件通用# my_package/my_module.py import json from confluent_kafka import Message def my_function(prefix: str, message: Message, threshold: int 0) - str | None: val json.loads(message.value()) if val[amount] threshold: return f{prefix}{val}函数对每条轮询到的消息应用一次返回真值时该值作为TriggerEvent的 payload否则继续轮询消息始终作为最后一个位置参数传入附加参数走apply_function_args/apply_function_kwargs对 Kafka 队列 Provider 来说apply_function必填Sensor 则允许传None此时以消息原始值UTF-8 解码作为事件 payload必须是点号字符串序列化限制且模块在 triggerer 中可导入。工作机制可概括为三步KafkaMessageQueueTrigger监听主题消息 →Asset抽象外部实体、AssetWatcher将 trigger 与 Asset 关联 → DAG 不再按固定调度运行而在 Asset 收到更新新消息时事件驱动执行。九、配置参考apache_kafka 与 kafka_event_producerProvider 注册了两个配置节定义于 get_provider_info.py L110-L2199.1[apache_kafka]Provider 公共设置选项类型默认值说明callback_allowliststring空逗号分隔的回调全路径白名单控制哪些error_cb/throttle_cb/stats_cb/log_cb/oauth_cb/on_commit字符串值可被解析执行。空 完全禁用字符串回调。托管认证不受影响2.0.0 新增9.2[kafka_event_producer]DagRun/TaskInstance 事件发布插件本包附带kafka_event_producer插件KafkaEventProducerPlugin通过 pyproject.toml 的airflow.pluginsentry point 注册将 Airflow 的 DagRun / TaskInstance 状态变更事件发布到 Kafka 主题。全部选项均自 1.14.1 起可用选项类型默认值说明dag_run_events_enabledbooleanFalse发布 DagRun 状态事件dag_run.running/success/failedFalse 时不注册 DagRun 监听器task_instance_events_enabledbooleanFalse发布 TaskInstance 状态事件task_instance.running/success/failed/skippedkafka_config_idstring构建 producer 所用 Airflow 连接未设置时回退默认连接kafka_defaulttopicstringairflow.events事件发布的目标主题主题必须预先存在插件不会自动创建sourcestring写入每条消息source字段的标识用于区分共享同一主题的多个 Airflow 安装未设置时回退到发出事件的组件主机名dag_run_dag_id_allowliststring逗号分隔的 glob 模式设置后仅对匹配的 dag_id 发出 DagRun 事件空 全部dag_run_dag_id_denyliststring匹配的 dag_id 被跳过deny 优先于 allowtask_instance_dag_id_allowliststring同上针对 TaskInstance 事件task_instance_dag_id_denyliststring同上deny 优先task_instance_task_id_allowliststring针对 task_id 的 glob 白名单与 dag_id 白名单叠加生效两者都需通过mapped task 共享同一 task_id一个模式可覆盖所有 map 索引task_instance_task_id_denyliststringtask_id 的 glob 黑名单deny 优先topic_check_timeoutinteger10每次主题存在性检查允许阻塞等待 broker 响应的秒数topic_check_retry_intervalinteger60主题检查失败后的重试间隔秒十、测试与验证路径仓库为该 Provider 提供了三层测试可用于验证配置与用法是否正确系统测试 DAGtests/system/apache/kafka/example_dag_hello_kafka.py 演示 produce/consume/sensor 全链路example_dag_event_listener.py、example_dag_message_queue_trigger.py 与 example_dag_kafka_message_queue_trigger.py 分别演示事件监听与消息队列触发。集成测试tests/integration/apache/kafka/对 consumer/producer/admin client hook、operator、trigger 的真实集群交互做验证。单元测试tests/unit/apache/kafka/覆盖 hook 配置构建、回调白名单拒绝逻辑、operator/sensor/trigger 的行为断言。小结apache-airflow-providers-apache-kafka2.0.0 以confluent-kafka为底座围绕一个 JSON 化的 Config Dict 连接提供了从管理AdminClient、生产/消费两个 Operator、等待消息两个可 defer Sensor 与两个 Trigger、到统一消息队列框架KafkaMessageQueueProviderMessageQueueTrigger的完整能力并内置回调白名单与 MSK/Google 托管 Kafka 的自动认证机制。开发时的三个关键注意点设置commit_cadence时务必显式关闭enable.auto.committrigger 类组件的apply_function只能传可导入的点号字符串自定义回调必须先进[apache_kafka] callback_allowlist白名单。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Airflow Kafka Provider 包深度指南：安装依赖、连接配置与 Hook/Operator/Sensor/Trigger 源码级解析
Apache Airflow Kafka Provider 包深度指南安装依赖、连接配置与 Hook/Operator/Sensor/Trigger 源码级解析【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本文以apache-airflow-providers-apache-kafka2.0.0 版本文档为核心系统讲解该 Provider 包的版本要求、依赖矩阵与安装方式并结合仓库源码深入解析 Kafka 连接的安全配置机制回调白名单、MSK IAM 自动认证、核心组件Hook、Operator、Sensor、Trigger、消息队列、事件插件的实现细节读完后可独立完成 Kafka Provider 的部署、连接配置与 DAG 实战开发。一、包概览apache.kafka Provider 是什么apache-airflow-providers-apache-kafka是 Apache Airflow 官方的 Apache Kafka Provider 发行包当前版本2.0.0负责让 Airflow 工作流与 Kafka 集群进行生产、消费、监听与事件驱动交互。该包的所有类都位于airflow.providers.apache.kafkaPython 包中通过标准 provider 机制注册到 Airflow。从 provider 元数据定义文件 可以确认这个包对外暴露了哪些组件类型组件类型模块Operatorsairflow.providers.apache.kafka.operators.consume、airflow.providers.apache.kafka.operators.produceHookshooks.base、hooks.client、hooks.consume、hooks.produceSensorairflow.providers.apache.kafka.sensors.kafkaTriggerstriggers.await_message、triggers.msg_queue消息队列 Providerairflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider插件kafka_event_producerKafkaEventProducerPluginAsset URI schemekafka://含 sanitize、create_asset 与 OpenLineage 转换器连接类型kafka默认 Hook 为KafkaBaseHook其中kafka://方案注册意味着 Airflow 的 Asset数据集机制可以直接用kafka://开头的 URI 表示一个 Kafka 主题并在 OpenLineage 数据血缘中转换。这些注册信息最终通过 pyproject.toml 中的 entry pointairflow.providers.apache.kafka.get_provider_info:get_provider_info被 Airflow 在启动时发现。官方文档的入口是 Kafka Provider 文档首页其下包含连接、Hook、Operator、消息队列、Sensor、Trigger 等使用指南以及 配置参考 与 Python API 参考。二、安装与版本要求2.1 安装方式在已有 Airflow 安装之上直接通过 pip 安装即可pip install apache-airflow-providers-apache-kafka如需源码安装可参考 从源码安装 Provider 的文档。2.2 最低版本与依赖矩阵该 Provider 发行包支持的最低 Apache Airflow 版本为2.11.0。完整依赖要求与 pyproject.toml 中dependencies一致PIP 包版本要求apache-airflow2.11.0apache-airflow-providers-common-compat1.12.0asgiref2.3.0Python 3.143.11.1Python 3.14confluent-kafka2.6.0Python 3.142.13.2Python 3.14可以看出该包要求 Python 3.10底层客户端库为confluent-kafkalibrdkafka 的 Python 绑定并且针对 Python 3.14 单独提高了asgiref与confluent-kafka的版本下限。2.3 跨 Provider 依赖Cross-provider dependencies以下依赖用于启用包的完整功能需要从 PyPI 安装对应的 provider 发行包可通过 extras 一次性装好pip install apache-airflow-providers-apache-kafka[common.messaging]依赖包Extra 名称apache-airflow-providers-common-messagingcommon.messagingapache-airflow-providers-googlegoogle2.4 可选依赖Optional extras这些 extras 安装可选的第三方库以启用额外功能从 PyPI 安装时使用pip install apache-airflow-providers-apache-kafka[google]Extra引入的依赖用途googleapache-airflow-providers-googleGoogle Managed Kafka 的自动 OAuth token 认证mskaws-msk-iam-sasl-signer-python1.0.1Amazon MSK IAM 自动认证common.messagingapache-airflow-providers-common-messaging2.0.0通用消息队列框架MessageQueueTrigger等官方发布的 sdist / wheel 包可在 Apache 官方下载站点获取并校验 sha512 与 asc 签名包名与版本号以当前发行版2.0.0为准。三、Kafka 连接配置Config Dict、回调白名单与托管认证3.1 连接类型与默认连接 IDKafka 连接类型基于confluent-kafka库extra字段是一个 JSON 可序列化的配置字典即 librdkafka 的全部配置项如bootstrap.servers、group.id、enable.auto.commit等。在 Airflow UI 中创建连接时host、port、schema、login、password字段会被隐藏extra字段会被重命名为Config Dict——这一 UI 行为正是由 KafkaBaseHook 的get_ui_field_behaviour方法返回的hidden_fields、relabeling与placeholders占位提示{bootstrap.servers: localhost:9092, group.id: my-group}驱动的。Kafka 连接配置界面所有 Kafka Hook 与 Operator 默认使用连接 IDkafka_default。这个默认连接极其简略只适合最基础的测试生产环境应创建自己的连接。最小有效配置的校验逻辑在源码中很明确bootstrap.servers键必须存在且有值否则直接抛出ValueError见 base.py 的_build_config。一个典型的连接extra示例也见于 系统测试示例 DAG 中通过AIRFLOW_CONN_*环境变量注入的连接{ bootstrap.servers: broker:9092, group.id: my-group, enable.auto.commit: false, auto.offset.reset: beginning }详细文档见 Kafka 连接指南。3.2 回调白名单字符串回调的安全边界连接extra中error_cb、throttle_cb、stats_cb、log_cb、oauth_cb、on_commit这六项 librdkafka 配置允许以点号路径字符串如module.callback_func形式提供Hook 会在构建客户端前将其解析为可调用对象。可解析的键名常量定义在 base.pyCALLBACK_CONFIG_KEYS (error_cb, throttle_cb, stats_cb, log_cb, oauth_cb, on_commit)从源码的_resolve_callbacks实现base.py L106-L145可以确认其安全策略回调只在完整导入路径模块 属性如my_company.kafka.auth.oauth_cb被列入[apache_kafka] callback_allowlist配置时才被导入执行条目必须与连接值精确匹配只写裸模块名如my_company.kafka.auth不会授权该模块内的任何可调用对象。白名单为空默认值时任何字符串形式的回调都会被拒绝并抛出ValueError同时记录 warning 日志。托管认证Amazon MSK IAM、Google Managed Kafka由 Hook 内部注入回调不经过白名单不受其影响。这解释了 连接文档中的警告该机制是为防止连接配置中的恶意回调被导入执行。使用自定义回调时需在 Airflow 配置中显式添加[apache_kafka] callback_allowlist my_company.kafka.auth.oauth_cb3.3 Amazon MSK IAM 自动认证对 Amazon MSK含 provisioned 与 serverless集群连接extra只需配置SASL_SSLOAUTHBEARER{ bootstrap.servers: boot-abcde1.c2.kafka-serverless.us-east-1.amazonaws.com:9098, security.protocol: SASL_SSL, sasl.mechanism: OAUTHBEARER, group.id: my-group }从 base.py 的_maybe_add_msk_iam_oauth可看到自动注入逻辑的完整判断链先用正则MSK_BOOTSTRAP_SERVERS_REGEXL43-L46识别 MSK 端点命名模式*.kafka.region.amazonaws.com与*.kafka-serverless.region.amazonaws.com含中国区.amazonaws.com.cn后缀并捕获 region仅当sasl.mechanism或sasl.mechanisms为OAUTHBEARER且匹配到 MSK 端点时才注入_msk_iam_oauth_cb(region)作为oauth_cb用户显式提供的oauth_cb永远优先不会被覆盖未安装mskextra 时抛出AirflowOptionalProviderFeatureException提示执行pip install apache-airflow-providers-apache-kafka[msk]AWS 凭证由 signer 通过标准 AWS 凭证链环境变量、共享配置文件、实例/任务 IAM 角色等解析。同理当bootstrap.servers指向 Google Managed Kafka包含cloud.goog与managedkafka字样时_build_config会自动导入 Google Provider 的ManagedKafkaHook并注入其get_confluent_token作为oauth_cbbase.py L159-L183需要预装googleProvider 14.1.0。值得注意的设计是test_connectionbase.py L230-L243调用同一个_build_config()来构建配置因此 UI 里“测试连接”与真实任务运行使用完全一致的回调解析与托管认证行为避免“UI 测试通过、任务却失败”的偏差。四、核心 HooksAdminClient、Producer、ConsumerHook 文档 列出了本包四个 HookHook模块说明KafkaBaseHook旧名KafkaHookhooks.base与 Kafka 交互的基类自定义 Kafka Hook 应继承它默认连接 ID 为kafka_defaultKafkaAdminClientHookhooks.client对 Kafka AdminClient 的封装用于集群管理操作KafkaConsumerHookhooks.consume创建 Kafka Consumer被ConsumeFromTopicOperator与AwaitMessageTrigger使用KafkaProducerHookhooks.produce创建 Kafka Producer被ProduceToTopicOperator使用从 base.py 可确认KafkaBaseHook的关键属性conn_type kafka、default_conn_name kafka_default、hook_name Apache Kafka其get_conn是一个cached_property返回以完整解析后配置构建的AdminClient。此外还有KafkaAuthenticationError自定义异常用于 Kafka 认证失败场景。五、OperatorsProduceToTopicOperator 与 ConsumeFromTopicOperatorOperator 文档 覆盖两个 Operator官方用法示例直接内嵌自 example_dag_hello_kafka.py5.1 ProduceToTopicOperator向主题生产消息t1 ProduceToTopicOperator( kafka_config_idt1-3, task_idproduce_to_topic, topictest_1, producer_functionexample_dag_hello_kafka.producer_function, )producer_function是用户提供的生成器/函数产出的 (key, value) 消息对会被发布到指定主题可以传点号字符串如上例DAG 序列化友好也可以直接传可调用对象。示例中对应的生成器def producer_function(): for i in range(20): yield (json.dumps(i), json.dumps(i 1))5.2 ConsumeFromTopicOperator批量消费并处理消息t2 ConsumeFromTopicOperator( kafka_config_idt2, task_idconsume_from_topic, topics[test_1], apply_functionexample_dag_hello_kafka.consumer_function, apply_function_kwargs{prefix: consumed:::}, commit_cadenceend_of_batch, max_messages10, max_batch_size2, )行为要点来自官方文档与示例 DAGOperator 创建一个 Kafka Consumer按批读取消息并用apply_function逐条处理或apply_function_batch整批处理直到到达日志末尾或达到max_messages上限提交语义若设置了commit_cadence参数必须确保连接配置中显式将enable.auto.commit设为false。默认enable.auto.commit为trueconsumer 每 5 秒自动提交 offset会覆盖commit_cadence定义的提交行为——这也是示例 DAG 中每个消费连接都显式带enable.auto.commit: False的原因设置return_apply_function_resultsTrue可返回apply_function每条消息返回的非None值列表按消费顺序返回值走普通 XCom 通道避免返回大负载该选项不适用于apply_function_batch。六、SensorsAwaitMessageSensor 与 AwaitMessageTriggerFunctionSensorSensor 文档 介绍两个可 defer 的 Sensor6.1 AwaitMessageSensor等待满足条件的消息t5 AwaitMessageSensor( kafka_config_idt5, task_idawaiting_message, topics[test_1], apply_functionexample_dag_hello_kafka.await_function, xcom_push_keyretrieved_message, )Sensor 会持续消费主题直到apply_function对某条消息返回真值此时触发TriggerEvent并成功完成。示例中的判断函数是“值能被 5 整除”def await_function(message): if json.loads(message.value()) % 5 0: return f Got the following message: {json.loads(message.value())}6.2 AwaitMessageTriggerFunctionSensor命中后触发回调再继续等待与上者不同该 Sensor 在消费到满足条件的消息后会调用event_triggered_function提供的可调用对象然后再次 defer 继续消费适合“事件监听器”类场景示例见 example_dag_event_listener.py。重要约束apply_function必须以点号字符串形式提供而不能传函数对象——因为 trigger 参数会被序列化进元数据库函数在triggerer 进程中被导入执行模块必须在那里可导入且修改函数需要重启 triggerer。详细要求见 消息队列文档中的 apply_function 一节。七、TriggersAwaitMessageTrigger 与 KafkaMessageQueueTriggerTrigger 文档 定义两个 triggerAwaitMessageTriggerairflow.providers.apache.kafka.triggers.await_message在 triggerer 中消费轮询到的 Kafka 消息并交给apply_function处理可调用返回任意数据时即 raiseTriggerEvent把任务唤醒。这是AwaitMessageSensor的运行时实现。KafkaMessageQueueTriggerairflow.providers.apache.kafka.triggers.msg_queueKafka 消息队列的专用接口类继承通用airflow.providers.common.messaging.triggers.msg_queue.MessageQueueTrigger与KafkaMessageQueueProvider配合提供 Kafka 消息队列操作的具体接口。八、消息队列集成KafkaMessageQueueProvider 与 MessageQueueTrigger本包还实现了 Kafka 消息队列 Providerairflow.providers.apache.kafka.queues.kafka.KafkaMessageQueueProvider是通用BaseMessageQueueProvider的 Kafka 后端允许在 Airflow 工作流中用 Kafka 主题作为消息队列并通过统一的MessageQueueTrigger接口收发。典型用法用 Kafka 主题上的新消息触发 DAGkafka://Asset 方案 监听器官方示例见 example_dag_kafka_message_queue_trigger.pyfrom airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger trigger MessageQueueTrigger( schemekafka, topics[my_topic], apply_functionmy_package.my_module.my_function, apply_function_args[received:], apply_function_kwargs{threshold: 100}, )apply_function的调用约定对触发器类组件通用# my_package/my_module.py import json from confluent_kafka import Message def my_function(prefix: str, message: Message, threshold: int 0) - str | None: val json.loads(message.value()) if val[amount] threshold: return f{prefix}{val}函数对每条轮询到的消息应用一次返回真值时该值作为TriggerEvent的 payload否则继续轮询消息始终作为最后一个位置参数传入附加参数走apply_function_args/apply_function_kwargs对 Kafka 队列 Provider 来说apply_function必填Sensor 则允许传None此时以消息原始值UTF-8 解码作为事件 payload必须是点号字符串序列化限制且模块在 triggerer 中可导入。工作机制可概括为三步KafkaMessageQueueTrigger监听主题消息 →Asset抽象外部实体、AssetWatcher将 trigger 与 Asset 关联 → DAG 不再按固定调度运行而在 Asset 收到更新新消息时事件驱动执行。九、配置参考apache_kafka 与 kafka_event_producerProvider 注册了两个配置节定义于 get_provider_info.py L110-L2199.1[apache_kafka]Provider 公共设置选项类型默认值说明callback_allowliststring空逗号分隔的回调全路径白名单控制哪些error_cb/throttle_cb/stats_cb/log_cb/oauth_cb/on_commit字符串值可被解析执行。空 完全禁用字符串回调。托管认证不受影响2.0.0 新增9.2[kafka_event_producer]DagRun/TaskInstance 事件发布插件本包附带kafka_event_producer插件KafkaEventProducerPlugin通过 pyproject.toml 的airflow.pluginsentry point 注册将 Airflow 的 DagRun / TaskInstance 状态变更事件发布到 Kafka 主题。全部选项均自 1.14.1 起可用选项类型默认值说明dag_run_events_enabledbooleanFalse发布 DagRun 状态事件dag_run.running/success/failedFalse 时不注册 DagRun 监听器task_instance_events_enabledbooleanFalse发布 TaskInstance 状态事件task_instance.running/success/failed/skippedkafka_config_idstring构建 producer 所用 Airflow 连接未设置时回退默认连接kafka_defaulttopicstringairflow.events事件发布的目标主题主题必须预先存在插件不会自动创建sourcestring写入每条消息source字段的标识用于区分共享同一主题的多个 Airflow 安装未设置时回退到发出事件的组件主机名dag_run_dag_id_allowliststring逗号分隔的 glob 模式设置后仅对匹配的 dag_id 发出 DagRun 事件空 全部dag_run_dag_id_denyliststring匹配的 dag_id 被跳过deny 优先于 allowtask_instance_dag_id_allowliststring同上针对 TaskInstance 事件task_instance_dag_id_denyliststring同上deny 优先task_instance_task_id_allowliststring针对 task_id 的 glob 白名单与 dag_id 白名单叠加生效两者都需通过mapped task 共享同一 task_id一个模式可覆盖所有 map 索引task_instance_task_id_denyliststringtask_id 的 glob 黑名单deny 优先topic_check_timeoutinteger10每次主题存在性检查允许阻塞等待 broker 响应的秒数topic_check_retry_intervalinteger60主题检查失败后的重试间隔秒十、测试与验证路径仓库为该 Provider 提供了三层测试可用于验证配置与用法是否正确系统测试 DAGtests/system/apache/kafka/example_dag_hello_kafka.py 演示 produce/consume/sensor 全链路example_dag_event_listener.py、example_dag_message_queue_trigger.py 与 example_dag_kafka_message_queue_trigger.py 分别演示事件监听与消息队列触发。集成测试tests/integration/apache/kafka/对 consumer/producer/admin client hook、operator、trigger 的真实集群交互做验证。单元测试tests/unit/apache/kafka/覆盖 hook 配置构建、回调白名单拒绝逻辑、operator/sensor/trigger 的行为断言。小结apache-airflow-providers-apache-kafka2.0.0 以confluent-kafka为底座围绕一个 JSON 化的 Config Dict 连接提供了从管理AdminClient、生产/消费两个 Operator、等待消息两个可 defer Sensor 与两个 Trigger、到统一消息队列框架KafkaMessageQueueProviderMessageQueueTrigger的完整能力并内置回调白名单与 MSK/Google 托管 Kafka 的自动认证机制。开发时的三个关键注意点设置commit_cadence时务必显式关闭enable.auto.committrigger 类组件的apply_function只能传可导入的点号字符串自定义回调必须先进[apache_kafka] callback_allowlist白名单。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考