Python Airflow 自定义传感器实现哪些方法

发布于08月27日

我正在try 编写一个继承BaseSensorOperator和SSHOperator的自定义传感器.在传感器主体中，我重写了poke和execute方法.

  class mysensor (BaseSensorOperator,SSHOperator):
      template_fields = ("my_param")
      def __init__ (
          self,
          my_param = None,
          *args,
          **kawags):
      self.my_param = my_param
      super().__init__(
          command = "some bash command",
          ssh_conn_id = "some connection"
          *args,
          **kwargs)

def poke(self, context):
    ... (implemented poke)
def execute():
    ...(implemented execute)

推荐答案

只需使用定制传感器即可实现poke功能.您不应该重写execute函数(除非您真的知道自己在做什么).执行功能在BaseSensorOperator中实现，这就是传感器的能力所在.

在你的情况下，你不应该使用SSHOperator，你应该直接使用SSHHook.

您的传感器应如下所示:

from airflow.providers.ssh.hooks.ssh import SSHHook

class Mysensor(BaseSensorOperator):
    ...

    def poke(self,context):
      hook = SSHHook(...)
      # add your sensor logic here