AWS托管Airflow使用aioredis时ZSET识别为字符串无法调用zrange方法

问题根因排查 #

• aioredis版本不一致

  本地运行环境安装的是aioredis v2.x及以上版本，该版本zadd方法支持传入{成员: 分数}的字典参数，和redis-py的接口规则一致。而AWS托管Apache Airflow v1.10.12的默认运行环境中aioredis为v1.x版本，该版本zadd的传参规则为zadd(键名, 分数, 成员)，直接传入字典会被错误序列化为字符串写入，直接导致对应key的类型变为string而非ZSET，这是该问题最高概率的触发原因。

• 其他任务误写同名key

  排查当前DAG的其他分支逻辑、或者其他运行中的DAG任务，是否对匹配matchingstring:*规则的key执行过set字符串写入操作，覆盖了你写入的ZSET类型数据。

• Redis连接配置偏差

  确认Airflow中配置的Redis连接对应的实例、数据库序号和本地测试环境完全一致，排除不同实例/不同库下存在同名string类型key的干扰。

修复方案 #

1. 统一依赖版本
   在AWS托管Airflow的依赖配置文件requirements.txt中指定和本地一致的aioredis版本，示例如下：

   aioredis==2.0.1
   arrow==1.2.3
   

   注意Airflow v1.10.12的运行环境Python版本为3.7，需确认指定的依赖版本适配Python 3.7。
2. 兼容低版本aioredis写法
   如果不希望升级依赖，可以直接调整zadd的参数写法适配aioredis v1.x的接口规则：
   原写入代码：
   redis.zadd(timestamp_key, {message.id: arrow.get(message.time).timestamp()})
   修改后代码：
   redis.zadd(timestamp_key, arrow.get(message.time).timestamp(), message.id)
3. 增加运行时校验逻辑
   调用zrangebyscore前先校验key类型，避免异常中断任务：

   key_type = redis.type(timestamp_key)
   if key_type == b'zset' or key_type == 'zset':
       res = redis.zrangebyscore(timestamp_key, min_value, max_value)
   else:
       # 自定义异常处理逻辑，例如日志上报、删除无效key后重试写入等
   

内容的提问来源于stack exchange，提问作者svitlanag