Provide shared storage between task via pluggable storage providers (akin to S3 remote logging)

[ecerulm]

Description

Provide shared storage between tasks.

Use case / motivation

Today we have XCom to share small pieces of information between task, but that is not suitable to share files.
Being able to share files to downstream task will simply new operators as you can create operators that produce files and operators that consume the files without having to implement a way to pass the files everytime.

Are you willing to submit a PR?

Yes,

Related Issues

[ecerulm]

My idea was to implement with with a pluggable storage so that S3, and other storage can be used.
My initial idea on how this could work is the following:

• When the task starts it downloads (if exists) the artifacts/dag_id/run_id/task_id/artifacts.zip for each upstream task
• Unzip all into local temporary artifacts/ directory allocated for the task
• run the operator, which have access to the artifacts/ directory (it gets the path with api call)
• when the operator finishes, the current status of the artifacts directory is packed as zip and uploaded to remote storage artifacts/dag_id/run_id/current_task_id/artifacts.zip

This is akin to how many CI/CD software (travisci, gitlabcicd) works, where the "artifacts" directory is automatically packed at the end of a job and unpacked at the start of the next job. The difference being that usually the storage is part of the CI/CD solution and here I'm proposing something similar to airflow's remote logging where the location of logs could be set to a remote location.

[vemikhaylov]

• When the task starts it downloads (if exists) the artifacts/dag_id/run_id/task_id/artifacts.zip for each upstream task

Will this be the default behaviour? If so, won't it be the excessive overhead leading to task start delay and network load? Or should we either explicitly state which artificats are required by the task?

[kaxil]

You can already do this via an XComBackend:

https://airflow.apache.org/blog/airflow-1.10.12/#allow-defining-custom-xcom-class

There is also an example in https://medium.com/apache-airflow/airflow-2-0-dag-authoring-redesigned-651edc397178

import pandas as pd
from typing import Any
from airflow.models.xcom import BaseXCom
from airflow.providers.google.cloud.hooks.gcs import GCSHook
class GCSXComBackend(BaseXCom):
   PREFIX = "xcom_gs://"
   BUCKET_NAME = "airflow-xcom-backend"
   @staticmethod
   def serialize_value(value: Any):
       if isinstance(value, pd.DataFrame):
           hook = GCSHook()
           object_name = "data_" + str(uuid.uuid4())
           with hook.provide_file_and_upload(
               bucket_name=GCSXComBackend.BUCKET_NAME,
               object_name=object_name,
           ) as f:
               value.to_csv(f.name)
           # Append prefix to persist information that the file
           # has to be downloaded from GCS
           value = GCSXComBackend.PREFIX + object_name
       return BaseXCom.serialize_value(value)
   @staticmethod
   def deserialize_value(result) -> Any:
       result = BaseXCom.deserialize_value(result)
       if isinstance(result, str) and result.startswith(GCSXComBackend.PREFIX):
           object_name = result.replace(GCSXComBackend.PREFIX, "")
           with GCSHook().provide_file(
               bucket_name=GCSXComBackend.BUCKET_NAME,
               object_name=object_name,
           ) as f:
               f.flush()
               result = pd.read_csv(f.name)
       return result

[xinbinhuang]

I think it would be nice to have CustomXComBackend for some common providers (e.g. GCP, AWS, Azure, etc.) added to core/providers as similar to SecretsBackend. What do you think?

[kaxil]

There is some discussion around it here:

• https://lists.apache.org/thread.html/rd6926b3442ad85f1ce72c5e2e80d8a31c0b5707da17b614ebb48c8c4%40%3Cdev.airflow.apache.org%3E
• WIP: RedisXCom backend #12733
• Added a second example for TaskFlow API #12170

Would love to get your input on those @xinbinhuang

[ecerulm]

I must confess that I didn't know about Airflow 2.0: DAG Authoring Redesigned and that it has been eye opening.

So I guess the approach in 2.x to pass data from one task/operator A to another task/operator B is:

• Return a value from the operator/task A. Operators have output(s) (for example you could return a pandas's DataFrame from PythonOperator)
• operator/task B can declare that they take some other operator/task B output as input with op_args=(taskA.output,)
• You provide an XComBackend that
  ** when task A returns a value takes care serializes it to a file, store in S3/etc, and then replaces the value that stored in XCom (which still has a limit of 48K) as maybe a string "pandasdf-s3://xxxx/mydata.parquet" and that is what really get stored in postgres and passed around by airflow
  ** when task B inputs are read (the value would be "pandas-df s3://xxx/mydate.parquet" the xcom backed will fetch the file from s3, deserialize into a pandas dataframe and pass that to the operator.

So I don't see any problem with that,but from the links that @kaxil posted I believe there no interest for a generic implementation XComBackend for S3, etc to be maintained in the airflow project, because since tasks returns arbitrary python objects there is no generic efficient way to serialize them, for example if the task return a pandas's DataFrame you don't want to use pickle to serialize you want to use df.to_parquet(...) instead.

So this feature request probably should be close but I think there are some follow up feature request that we can further discuss:

• Maybe a another issue that advocates for creating a S3XComBackendBase that serve as a base to implement custom xcom backends just by providing the serialization logic. As I think part of the problem is that the serialize and the remote storage is conflated.
• maybe the possibility of even providing an XComBackend per DAG , or at least a serializer per DAG, so that the guy authoring the DAG can decide the best way of serializing each type of value

so @kaxil where is the best place to continue the discussion, in the mailing list or here?

[kaxil]

@ecerulm Mailing list is the best place

[potiuk]

This is really part of several AIPs that have been raised in the past:

• https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-5+Remote+DAG+Fetcher
• https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-20+DAG+manifest

And those should be possibly picked up and lead to completion.

There are also possible improvements to XCom backends possibly but also those shoudl be discussed in mailing list as Kaxil mentioned.

Converting to discussion in case more discussion is needed.