这是一种有点自以为是的做法。
正确的方法是首先创建一个hook，然后是operator，它将使用这个钩子。对于以下更简单的情况，您甚至不需要在运算符中调用钩子。
#1。 放置
<PROJECT NAME>/<PLUGINS_FOLDER>/<PLUGIN NAME>/__init__.py
<PROJECT NAME>/<PLUGINS_FOLDER>/<PLUGIN NAME>/<some_new>_hook.py
<PROJECT NAME>/<PLUGINS_FOLDER>/<PLUGIN NAME>/<some_new>_operator.py
对于看起来像这样的真实案例：
CRMProject/crm_plugin/__init__.py
CRMProject/crm_plugin/crm_hook.py
CRMProject/crm_plugin/customer_operator.py
#2。 代码
CRMProject/crm_plugin/__init__.py的示例代码：
# CRMProject/crm_plugin/__init__.py
from airflow.plugins_manager import AirflowPlugin
from crm_plugin.crm_hook import CrmHook
from crm_plugin.customer_operator import CreateCustomerOperator, DeleteCustomerOperator, UpdateCustomerOperator
class AirflowCrmPlugin(AirflowPlugin):
name = "crm_plugin" # does not need to match the package name
operators = [CreateCustomerOperator, DeleteCustomerOperator, UpdateCustomerOperator]
sensors = []
hooks = [CrmHook]
executors = []
macros = []
admin_views = []
flask_blueprints = []
menu_links = []
appbuilder_views = []
appbuilder_menu_items = []
global_operator_extra_links = []
operator_extra_links = []
钩子类的示例代码 - CRMProject/crm_plugin/crm_hook.py。永远不要直接从 system\API 调用它。为此使用运算符（见下文）。
from airflow.hooks.base_hook import BaseHook
from airflow.exceptions import AirflowException
from crm_sdk import crm_api # import external libraries to interact with target system
class CrmHook(BaseHook):
"""
Hook to interact with the ACME CRM System.
"""
def __init__(self, ...):
# your code goes here
def insert_object(self, ...):
"""
Insert an object into the CRM system
"""
# your code goes here
def update_object(self, ...):
"""
Update an object into the CRM system
"""
# your code goes here
def delete_object(self, ...):
"""
Delete an object into the CRM system
"""
# your code goes here
def extract_object(self, ...):
"""
Extract an object into the CRM system
"""
# your code goes here
您将在 DAG 中使用的运算符 (CRMProject/crm_plugin/customer_operator.py) 的示例代码。运算符要求您实现一个执行方法。这是 Airflow 操作员的入口点，当 DAG 中的任务执行时会调用它。
apply_defaults 装饰器包装了类的 __init__ 方法，该方法将在 DAG 脚本中设置的 DAG 默认值应用于运行时操作员的任务实例。
我们还可以设置两个重要的类属性。它们是templated_fields 和template_ext。这两个属性是可迭代的，应该包含字段和/或文件扩展名的字符串值，这将允许使用 Airflow 中的 jinja 模板支持进行模板化。
from airflow.exceptions import AirflowException
from airflow.operators import BaseOperator
from airflow.utils.decorators import apply_defauls
from crm_plugin.crm_hook import CrmHook
class CreateCustomerOperator(BaseOperator):
"""
This operator creates a new customer in the ACME CRM System.
"""
template_fields = ['first_contact_date', 'bulk_file']
template_ext = ['.csv']
@apply_defaults
def __init__(self, first_contact_date, bulk_file, ...):
# your code goes here
def _customer_exist(self, ...):
"""
Helper method to check if a customer exist. Raises an exception if it does.
"""
# your code goes here
def execute(self, context):
"""
Create a new customer in the CRM system.
"""
# your code goes here
您可以根据需要在类中创建任意数量的方法，以简化执行方法。良好类设计的相同原则在这里仍然很重要。
#3。 部署和使用您的插件
完成插件工作后，您只需将 <PLUGIN NAME> 包文件夹复制到 Airflow 插件文件夹即可。 Airflow 将选择该插件，并且它将可供您的 DAG 使用。
如果我们将简单的 CRM 插件复制到我们的 plugins_folder，文件夹结构将如下所示。
<plugins_folder>/crm_plugin/__init__.py
<plugins_folder>/crm_plugin/crm_hook.py
<plugins_folder>/crm_plugin/customer_operator.py
为了使用您的新插件，您只需使用以下语句导入您的运算符和挂钩。
from airflow.hooks.crm_plugin import CrmHook
from airflow.operators.crm_plugin import CreateCustomerOperator, DeleteCustomerOperator, UpdateCustomerOperator
Source