我正在尝试实现自定义的XCOM后端。
以下是我所做的步骤:
- 在主Airflow目录(AIRFLOW_HOME)下创建了“include”目录。
- 在其中创建了名为“custom_xcom_backend.py”的文件。
from typing import Any
from airflow.models.xcom import BaseXCom
import pandas as pd
class CustomXComBackend(BaseXCom):
@staticmethod
def serialize_value(value: Any):
if isinstance(value, pd.DataFrame):
value = value.to_json(orient='records')
return BaseXCom.serialize_value(value)
@staticmethod
def deserialize_value(result) -> Any:
result = BaseXCom.deserialize_value(result)
result = df = pd.read_json(result)
return result
在配置文件中设置:
- 在配置文件中进行设置:
xcom_backend = include.custom_xcom_backend.CustomXComBackend
当我重新启动Web服务器时,我遇到了以下问题:
airflow.exceptions.AirflowConfigException: The object could not be loaded. Please check "xcom_backend" key in "core" section. Current value: "include.cust...
我猜测它没有识别到“include”文件夹,但我该如何修复呢?*注意:没有使用docker,而是安装在Ubuntu机器上。谢谢!