【免费下载链接】context-hub项目地址https://gitcode.com/gh_mirrors/co/context-hub点击查看免费下载本文基于 Context Hub 仓库中的 Apache Airflow Redis Provider 指南content/apache-airflow/docs/providers-redis/python/DOC.md对应apache-airflow-providers-redis4.4.2编写。它面向这样一个场景当你的 Airflow DAG 需要通过 Airflow 连接Connection访问 Redis、在 Python 任务代码中执行 Redis 命令或者向 Redis 频道发布消息时本指南给出从安装、连接配置到任务编写的完整实操路径。读完本文你将掌握用RedisHook在任务内获得 Redis 客户端、用RedisPublishOperator一步完成频道发布以及把连接凭据安全收敛到 Airflow 连接层的标准做法。这份文档以docs类型收录在 Context Hub 中详见 Content Guide 的目录与 frontmatter 规范Agent 可通过chub search/chub get直接拉取使用参考 CLI Reference 与 get-api-docs skill确保写代码时读到的是与当前版本匹配的最新说明。黄金法则Provider 不是独立的 Redis 客户端使用apache-airflow-providers-redis之前先建立三个核心认知必须与apache-airflow一起安装。它不是一个独立的 Redis 客户端或 worker 运行时而是 Airflow 生态里的 provider 包依赖 Airflow 的 Hook / Operator 机制运行。把 Redis 的主机、端口、密码和数据库选择放在 Airflow 连接上让 DAG 代码只关注 key、channel 和任务逻辑而不是连接细节。按职责选择入口任务代码需要直接执行 Redis 命令时用RedisHook任务只是“把这条消息发布到这个频道”时用RedisPublishOperator。凭据一律放在 Airflow 连接或 secrets backend 中严禁硬编码在 DAG 文件里。这个包提供了什么该 provider 为 Airflow 提供 Redis 集成主要入口有两个也是绝大多数 DAG 会用到的东西入口模块路径用途RedisHookairflow.providers.redis.hooks.redis在任务代码中获得可直接调用的 Redis 客户端RedisPublishOperatorairflow.providers.redis.operators.redis_publish以独立任务形式向指定频道发布一条消息注意Airflow Provider 包与 Airflow 核心版本各自独立演进二者互不绑定升级任何一方前都需要核对兼容性。安装将 provider 安装到 Airflow 部署所用的同一个 Python 环境或容器镜像中。遵循 Airflow 常规的 provider 安装方式并且在同一条命令里固定 Airflow 版本避免pip静默地把 core 升级到不兼容的版本。python -m venv .venv source .venv/bin/activate python -m pip install --upgrade pip AIRFLOW_VERSIONyour-airflow-version PROVIDER_VERSION4.4.2 PYTHON_VERSION$(python -c import sys; print(f{sys.version_info.major}.{sys.version_info.minor})) # 使用 Airflow 官方 constraints 文件格式为 constraints-airflow-version/constraints-python-version.txt CONSTRAINT_URLairflow-constraints-url-for-your-python-version python -m pip install \ apache-airflow${AIRFLOW_VERSION} \ apache-airflow-providers-redis${PROVIDER_VERSION} \ --constraint ${CONSTRAINT_URL}安装后建议做一次验证airflow providers list | grep -i redis也可以直接用 Python 验证导入路径可用python -c from airflow.providers.redis.hooks.redis import RedisHook; from airflow.providers.redis.operators.redis_publish import RedisPublishOperator; print(ok)实际部署中凡是会导入 DAG 或执行任务的镜像/环境都必须安装该 provider——只装在 webserver 或 scheduler 上worker 上缺包会导致任务运行时导入失败。配置 Redis 连接创建 connection type 为redis的 Airflow 连接或直接用环境变量定义。环境变量方式最简洁export AIRFLOW_CONN_REDIS_DEFAULTredis://:secretredis.example.com:6379/0设置该环境变量后约定如下Airflow 连接 id 为redis_defaultRedisHook默认使用redis_conn_idredis_defaultRedisPublishOperator同样默认使用redis_conn_idredis_default。实用要点URI 末尾的/0路径段用于选择 Redis 数据库这里表示数据库0密码中的特殊字符必须做 URL 编码后再放进AIRFLOW_CONN_REDIS_DEFAULT如果不希望用 URI 形式也可以在 Airflow UI 或 CLI 中以等价的主机、端口、密码、数据库设置来创建连接。凭证应始终保存在 Airflow 连接或 secrets backend 中而不是嵌入任务代码。在任务中使用 RedisHook当任务代码需要直接访问 Redis 客户端时使用RedisHook。下面是一个完整的示例 DAGfrom airflow import DAG from airflow.decorators import task from airflow.providers.redis.hooks.redis import RedisHook from pendulum import datetime with DAG( dag_idredis_hook_example, start_datedatetime(2026, 1, 1), scheduleNone, catchupFalse, ) as dag: task def write_and_read() - str | None: hook RedisHook(redis_conn_idredis_default) client hook.get_conn() if not client.ping(): raise RuntimeError(Redis is not reachable) client.set(airflow:demo:status, ready, ex300) value client.get(airflow:demo:status) return value.decode(utf-8) if value else None write_and_read()关键模式只有三步创建RedisHook(redis_conn_id...)调用hook.get_conn()获取底层 Redis 客户端用返回的客户端执行常规 Redis 命令例如ping()、set()、get()、delete()、publish()。几个值得注意的细节参数名是redis_conn_id不是conn_id或redis_id——这是最常见的写错点set(key, value, ex300)中的ex指定键的过期秒数示例中 300 秒后键自动失效Redis 客户端返回的get()结果是字节串bytes需要value.decode(utf-8)转成字符串空值None要单独处理ping()是连通性检查的首选失败时抛RuntimeError能让任务快速失败、便于排障。用 RedisPublishOperator 发布消息当任务仅仅是“向某频道发布消息”、不需要围绕它写自定义 Python 代码时使用RedisPublishOperatorfrom airflow import DAG from airflow.providers.redis.operators.redis_publish import RedisPublishOperator from pendulum import datetime with DAG( dag_idredis_publish_example, start_datedatetime(2026, 1, 1), scheduleNone, catchupFalse, ) as dag: publish_event RedisPublishOperator( task_idpublish_event, redis_conn_idredis_default, channelevents, message{type: order.created, id: ord_123}, )这是从 DAG 发出 pub/sub 通知最简单直接的写法。参数含义参数说明task_id任务唯一标识redis_conn_id指向 Airflow 中 Redis 连接的 idchannel要发布到的 Redis 频道名如eventsmessage发布的消息内容可以是任意字符串实践中常用 JSON 文本消息发布后任何订阅了该频道的 Redis 客户端其他服务、另一个 DAG 的监听任务等都会收到通知。这种“事件广播”模式非常适合把订单创建、任务完成等业务事件从 DAG 中异步地推送给下游消费者。常见设置模式对大多数 DAG 而言一个干净的分层是连接层Redis 端点与凭据放在 Airflow 连接中不进 DAG 源码命令层需要执行常规 Redis 命令时在任务内部通过RedisHook创建客户端发布层只需一步发布时用RedisPublishOperator作为独立任务返回值任务只返回小的派生值避免返回大 payload 或大 key 的全量 dump以免撑爆 XCom。常见陷阱只装在一部分组件上provider 只装到 webserver 或 scheduler而 worker 上任务代码会import airflow.providers.redis——任何执行任务的运行时缺包都会导入失败连接参数名写错RedisHook与RedisPublishOperator用的都是redis_conn_id把密码或主机名直接写进 DAG 代码应使用 Airflow 连接或 secrets backend忘记对保留字符做 URL 编码手动定义AIRFLOW_CONN_REDIS_DEFAULT时密码中的、:、/等字符必须编码假设连接只对某个容器有效某个 Airflow 容器能连通的 Redis不代表所有 worker 都能连通务必确认每个运行时都能实际访问到 Redis 端点。版本说明本指南覆盖apache-airflow-providers-redis4.4.2升级 provider 时先在 Airflow 镜像中验证兼容性再滚动上线并重新核对 provider 文档与 API reference。延伸阅读本文对应的原始文档content/apache-airflow/docs/providers-redis/python/DOC.mdfrontmatter 标注语言为 Python、版本 4.4.2、revision 1、source: maintainer本仓库还收录了 Redis 官方 Python 客户端文档 content/redis/docs/key-value/python/DOC.md可配合理解 Hook 底层所调用的 Redis 命令语义Context Hub 的文档组织与 frontmatter 规范见 Content GuideCLI 用法见 CLI ReferenceAgent 自动取文档的流程见 get-api-docs skill。赞分享【免费下载链接】context-hub项目地址https://gitcode.com/gh_mirrors/co/context-hub点击查看免费下载相关推荐Apache Airflow Plexus Provider 实战指南基于 PlexusHook 与 PlexusJobOperator 在 DAG 中完成 Plexus 作业提交Apache Airflow Plexus Provider 实战指南基于 PlexusHook 与 PlexusJobOperator 在 DAG 中完成Apache Storm 与 Redis 集成实战指南基于 storm-redis 的 Bolt 与 Trident State 开发Apache Storm 与 Redis 集成实战指南基于 storm redis 的 Bolt 与 Trident State 开发 导读 本文以 Apac流处理后端大数据Apache Airflow SFTP Provider 实战指南用 SFTPHook 与 SFTPOperator 构建 DAG 文件传输任务Apache Airflow SFTP Provider 实战指南用 SFTPHook 与 SFTPOperator 构建 DAG 文件传输任务 本文是基于上一篇CANN ops-nn 算子详解GatherElements 索引收集算子功能、实现与适配指南下一篇Draggable 的 TouchSensor 深入解析移动端触摸拖拽传感器的工作原理与配置实战创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考