Portable 插件 Python SDK

用户可利用 Python SDK 来开发 portable 插件，这个 SDK 提供了类似原生插件的 API，另外它提供了启动函数，用户只需填充插件信息即可。

为了运行 python 插件，有两个前置条件 To run python plugin, there are two prerequisites in the runtime environment:

  1. 安装 Python 3.x 环境.
  2. 通过 pip install nng ekuiper 安装 nng 和 ekuiper 包.

默认情况下，eKuiper 的 portable 插件运行时会通过 python 命令来运行插件。如果您的环境不支持 python 命令，请通过配置文件更换为可用的 Python 命令。

插件开发

开发插件包括子模块和主程序两部分, Python SDK 提供了 python 语言的源，目标和函数 API。

源接口:

  1.   class Source(object):
  2.     """abstract class for eKuiper source plugin"""
  4.     @abstractmethod
  5.     def configure(self, datasource: str, conf: dict):
  6.         """configure with the string datasource and conf map and raise error if any"""
  7.         pass
  9.     @abstractmethod
  10.     def open(self, ctx: Context):
  11.         """run continuously and send out the data or error with ctx"""
  12.         pass
  14.     @abstractmethod
  15.     def close(self, ctx: Context):
  16.         """stop running and clean up"""
  17.         pass

目标接口:

  1. class Sink(object):
  2.     """abstract class for eKuiper sink plugin"""
  4.     @abstractmethod
  5.     def configure(self, conf: dict):
  6.         """configure with conf map and raise error if any"""
  7.         pass
  9.     @abstractmethod
  10.     def open(self, ctx: Context):
  11.         """open connection and wait to receive data"""
  12.         pass
  14.     @abstractmethod
  15.     def collect(self, ctx: Context, data: Any):
  16.         """callback to deal with received data"""
  17.         pass
  19.     @abstractmethod
  20.     def close(self, ctx: Context):
  21.         """stop running and clean up"""
  22.         pass

函数接口:

  1. class Function(object):
  2.     """abstract class for eKuiper function plugin"""
  4.     @abstractmethod
  5.     def validate(self, args: List[Any]):
  6.         """callback to validate against ast args, return a string error or empty string"""
  7.         pass
  9.     @abstractmethod
  10.     def exec(self, args: List[Any], ctx: Context) -> Any:
  11.         """callback to do execution, return result"""
  12.         pass
  14.     @abstractmethod
  15.     def is_aggregate(self):
  16.         """callback to check if function is for aggregation, return bool"""
  17.         pass

用户通过实现这些抽象接口来创建自己的源，目标和函数，然后在主函数中声明这些自定义插件的实例化方法

  1. if __name__ == '__main__':
  2.     c = PluginConfig("pysam", {"pyjson": lambda: PyJson()}, {"print": lambda: PrintSink()},
  3.                      {"revert": lambda: revertIns})
  4.     plugin.start(c)

关于更详细的信息，请参考这篇文章 python sdk example.

打包发布

由于 python 是解释性语言，不需要编译出可执行文件，需要确保 json 描述文件中可执行文件名字的准确性即可。详细信息，请参考

部署要求

运行 python 脚本需要有 python 环境。所以，目标系统必须安装 python 3.x 环境。如果使用 docker ，建议使用 lfedge/ekuiper:<tag>-slim-python 版本。该版本包含 eKuiper 和 python 环境，无需再手动安装。