Airflow - Custom XCom backend on Ubuntu

Viewed 288

I'm trying to implement custom XCOM backend.

Those are the steps I did:

  1. Created "include" directory at the main Airflow dir (AIRFLOW_HOME).
  2. Created these "custom_xcom_backend.py" file inside:
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

  1. Set at config file:
xcom_backend = include.custom_xcom_backend.CustomXComBackend
  1. When I restarted webserver I got:
airflow.exceptions.AirflowConfigException: The object could not be loaded. Please check "xcom_backend" key in "core" section. Current value: "include.cust...

My guess is that it not recognizing the "include" folder

But how can I fix it?

*Note: There is no docker. It is installed on a Ubuntu machine.

Thanks!

1 Answers

So I solved it:

  1. Put custom_xcom_backend.py into the plugins directory
  2. set at config file:
xcom_backend = custom_xcom_backend.CustomXComBackend
  1. Restart all airflow related services

*Note: Do not store DataFrames that way (bad practice).

Sources I used: https://www.youtube.com/watch?v=iI0ymwOij88

Related