编辑“︁Airflow自定义Sensor”︁（章节）

== 创建自定义Sensor ==
以下示例实现检查HDFS文件是否存在的Sensor：

<syntaxhighlight lang="python">
from airflow.sensors.base import BaseSensorOperator
from airflow.providers.apache.hdfs.hooks.hdfs import HDFSHook

class HdfsFileSensor(BaseSensorOperator):
    """
    自定义HDFS文件检查Sensor
    :param hdfs_conn_id: HDFS连接ID
    :param filepath: 要检查的文件路径
    """
    template_fields = ('filepath',)

    def __init__(self, hdfs_conn_id='hdfs_default', filepath=None, **kwargs):
        super().__init__(**kwargs)
        self.hdfs_conn_id = hdfs_conn_id
        self.filepath = filepath

    def poke(self, context):
        hook = HDFSHook(self.hdfs_conn_id)
        client = hook.get_conn()
        self.log.info(f'Checking HDFS path: {self.filepath}')
        return client.status(self.filepath, strict=False) is not None
</syntaxhighlight>

'''参数说明：'''
* <code>template_fields</code>：支持宏替换的字段
* <code>strict=False</code>：文件不存在时不抛出异常