Airflow中使用HDFS Sensor遇snakebite依赖Python3语法错误求方案

这个问题我之前帮不少人排查过，本质是snakebite库只兼容Python2——它代码里用了Python2特有的1L（长整数）写法，而Python3里已经统一用1表示整数，所以在Python3环境加载这个库就会触发语法错误。下面给你几个不用切换到Python2的可行方案，按推荐程度排序：

方案1：改用Airflow官方支持Python3的HDFS Provider（最推荐） #

如果你用的是Airflow 2.x版本，官方已经把HDFS相关功能迁移到了独立的provider包，完全适配Python3，而且不再依赖snakebite：

1. 安装对应的provider包：

   pip install apache-airflow-providers-apache-hdfs
   

2. 修改代码中的导入语句：
   把S3HDFSObjectSensor.py里的

   from airflow.hooks.hdfs_hook import HDFSHook
   

   替换成

   from airflow.providers.apache.hdfs.hooks.hdfs import HDFSHook
   

3. 适配Sensor逻辑：新的HDFSHook使用hdfs库作为客户端，API和旧版略有不同（比如检查文件存在用check_for_path()方法），你需要调整S3HDFSObjectSensor里的HDFS操作代码，按需适配即可。

方案2：使用Python3兼容的snakebite分支 #

如果因为历史依赖必须用snakebite，可以用社区维护的Python3兼容版本snakebite3：

1. 卸载原有的snakebite：

   pip uninstall -y snakebite
   

2. 安装snakebite3：

   pip install snakebite3
   

   这个分支已经把所有Python2专属语法（比如1L）改成了Python3兼容写法，安装后你的原有代码不需要大改就能正常运行。

方案3：自定义HDFSHook使用pyhdfs（灵活定制场景） #

如果上面两种方案都不满足你的需求，可以自己实现一个基于pyhdfs的HDFS钩子，pyhdfs是完全支持Python3的轻量HDFS客户端：

1. 安装pyhdfs：

   pip install pyhdfs
   

2. 编写自定义Hook（比如放在你的plugins目录下）：

   from airflow.hooks.base_hook import BaseHook
   import pyhdfs
   
   class PyHDFSHook(BaseHook):
       def __init__(self, hdfs_conn_id='hdfs_default'):
           super().__init__()
           self.conn_id = hdfs_conn_id
           self.client = self.get_conn()
   
       def get_conn(self):
           conn = self.get_connection(self.conn_id)
           # 根据你的HDFS集群配置调整参数，比如是否需要Kerberos认证
           return pyhdfs.HdfsClient(hosts=f"{conn.host}:{conn.port}", user_name=conn.login)
   
       def check_file_exists(self, path):
           return self.client.exists(path)
   

3. 在S3HDFSObjectSensor里导入这个自定义Hook，替换原有HDFSHook的调用逻辑即可。

另外，你的DAG代码本身不需要大改，只要底层的Hook能正常工作，S3HDFSObjectSensor的实例化代码可以保持原样。

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