Airflow v1.9.0自定义传感器报AttributeError：NoneType无get_records属性

看起来你遇到的问题是仅单个Celery Worker执行自定义传感器时报错，其他Worker和Scheduler CLI运行正常，错误指向get_records调用在None对象上。结合你的环境（Airflow v1.9.0 + Celery Executor）和已排查的点，我帮你梳理最可能的原因和解决方案：

1. 出错Worker缺失数据库驱动依赖 #

这是最大概率的原因：

• 你的自定义传感器通过BaseHook.get_connection(conn_id).get_hook()获取数据库操作句柄，而不同类型的数据库（比如Oracle、MySQL、PostgreSQL等）需要对应的Python驱动包才能生成有效的Hook实例。
• 如果这个出问题的Worker没有安装CCUSER_SOURCE_GWPROD_RPT或RDS_DMUAT_DMCONFIG连接对应的数据库驱动，get_hook()就会返回None，后续调用get_records自然会触发AttributeError。
• 验证方法：登录到出问题的Worker节点，进入Airflow使用的Python环境，执行以下测试代码：

  from airflow.hooks.base_hook import BaseHook
  # 测试源数据库连接
  conn_src = BaseHook.get_connection("CCUSER_SOURCE_GWPROD_RPT")
  hook_src = conn_src.get_hook()
  print(f"源库Hook: {hook_src}")
  # 测试目标数据库连接
  conn_tgt = BaseHook.get_connection("RDS_DMUAT_DMCONFIG")
  hook_tgt = conn_tgt.get_hook()
  print(f"目标库Hook: {hook_tgt}")
  

  如果输出是None，就说明确实缺少对应驱动，安装即可（比如Oracle装cx_Oracle，MySQL装mysqlclient）。

2. 出错Worker的元数据库连接配置异常 #

• 检查该Worker的airflow.cfg文件中sql_alchemy_conn配置，确认和其他正常Worker、Scheduler的配置完全一致。如果这个Worker连接的是另一个元数据库，里面没有你定义的两个conn_id，虽然BaseHook.get_connection通常会抛出AirflowNotFoundException，但不排除缓存或配置同步问题导致返回None。
• 也可以直接在Worker上执行airflow connections list命令，查看是否能列出CCUSER_SOURCE_GWPROD_RPT和RDS_DMUAT_DMCONFIG这两个连接，确认元数据库连接正常。

3. 自定义传感器代码的小瑕疵（非直接报错原因，但建议修复） #

你的传感器继承自SqlSensor，但__init__方法的super调用有问题：

# 你当前的写法
super(SqlSensor, self).__init__(*args, **kwargs)
# 正确的写法应该是
super(SensorWatermarkDt, self).__init__(*args, **kwargs)

当前写法会跳过SqlSensor自身的初始化逻辑，直接调用它的父类BaseSensorOperator的初始化。虽然这不是导致单个Worker报错的直接原因，但可能引发其他潜在问题，建议修复以保证代码规范。

4. 插件在该Worker上未正确加载 #

• 虽然你说重启了Worker，但可能该Worker的插件目录存在权限问题，或者插件文件没有同步到该节点。
• 检查该Worker的airflow.cfg中plugins_folder配置的路径，确认目录下存在你的PDIPlugin插件文件，且Airflow进程有读取权限。
• 查看Worker的启动日志，搜索PDIPlugin相关的加载日志，确认插件是否成功被加载。

优先排查顺序 #

1. 先验证Worker的数据库驱动是否缺失（最可能）；
2. 确认Worker的元数据库连接和连接配置正常；
3. 最后检查插件加载和代码继承问题。

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