You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/11/07 10:44:08 UTC
[GitHub] [airflow] bolkedebruin opened a new pull request, #27540: Allow datasets to be used in taskflow
bolkedebruin opened a new pull request, #27540:
URL: https://github.com/apache/airflow/pull/27540
Datasets could not be passed as parameters to taskflow functions as they could not be serialized. This commit:
1) changes the xcom serializer so that it now can serialize objects that have attr
2) removes the need to for a custom serializer in lineage
3) adds a version check to the serializer/deserializer
4) registers any datasets as either inlets or outlets in the task
Currently the deserializer is NOT compatible with the previous one used in lineage for the following reasons:
1) '__type' was overloaded (used as the encoding type AND actual type), the standardized deserializer can't figure out if it is seeing the actual type or encoded type as it receives the nested version first
2) '__source' is omitted, it seemed not to be used (could be wrong)
@ashb @jedcunningham @kaxil
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] BasPH commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
BasPH commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025180542
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
Review Comment:
Did some rewording for clarity
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020086076
##########
tests/utils/test_json.py:
##########
@@ -25,9 +25,26 @@
import pendulum
import pytest
+from airflow.datasets import Dataset
from airflow.utils import json as utils_json
+class Z:
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict):
+ return Z(data["x"])
Review Comment:
I wonder if we should call these `airflow_serialize` etc, to reduce the risk of these clashing when used with "existing" objects?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020082192
##########
airflow/utils/json.py:
##########
@@ -40,6 +43,16 @@
log = logging.getLogger(__name__)
+CLASSNAME = "__classname__"
+VERSION = "__version__"
+DATA = "__data__"
Review Comment:
Rather than creating yet another way of encoding this kind of thing in the code base, is it possible we could reuse the `__var` and `__type` keys/prefixes that the BaseSerialization uses, or would that cause problems?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020350738
##########
tests/utils/test_json.py:
##########
@@ -25,9 +25,26 @@
import pendulum
import pytest
+from airflow.datasets import Dataset
from airflow.utils import json as utils_json
+class Z:
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict):
+ return Z(data["x"])
Review Comment:
My idea is that serialization isn't something special and not necessarily tied to Airflow. In essence it is just a to_dict()/from_dict() which goes for anything in python.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020132940
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
+ }
+ elif attr.has(o.__class__):
+ classname = o.__module__ + "." + o.__class__.__name__
+
+ version = getattr(o, "version", DEFAULT_VERSION)
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: BaseSerialization.serialize(data),
+ }
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(object_hook=self.object_hook, *args, **kwargs)
+
+ def object_hook(self, dct) -> object:
+ dct = self._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if VERSION in dct and int(dct[VERSION]) < version:
+ raise TypeError(
+ "serialized version of %s is newer than module version (%s > %s)",
+ dct[CLASSNAME],
+ dct[VERSION],
+ version,
+ )
+
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA])
Review Comment:
yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] nick4fake commented on pull request #27540: Allow datasets to be used in taskflow
Posted by "nick4fake (via GitHub)" <gi...@apache.org>.
nick4fake commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1485188789
@ashb sorry, but will it work for scheduling?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025174245
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
+
+.. code-block:: python
+
+ from typing import ClassVar
+
+
+ class MyCustom:
+ version: ClassVar[int] = 1
+
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict, version: int):
+ if version > 1:
+ raise TypeError(f"version > {MyCustom.version}")
+ return MyCustom(data["x"])
Review Comment:
Yes, but I am not ready for that yet :-). Example added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] BasPH commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
BasPH commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025202077
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
+
+.. code-block:: python
+
+ from typing import ClassVar
+
+
+ class MyCustom:
+ version: ClassVar[int] = 1
+
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict, version: int):
+ if version > 1:
+ raise TypeError(f"version > {MyCustom.version}")
+ return MyCustom(data["x"])
Review Comment:
I meant an example using this custom class. I'm not reading anything in relation to datasets. As a user, what do I do with `MyCustom`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1017893141
##########
airflow/decorators/base.py:
##########
@@ -210,17 +212,27 @@ def __init__(
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
def execute(self, context: Context):
+ for arg in chain(self.op_args, self.op_kwargs.values()):
+ if isinstance(arg, Dataset):
+ self.inlets.append(arg)
Review Comment:
Are you suggesting that we would make this more generic? It now only affects the `DecoratedOperator` moving it to prepare_lineage (called from `pre_execute`) would make it affect all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025108730
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
+
+.. code-block:: python
+
+ from typing import ClassVar
+
+
+ class MyCustom:
+ version: ClassVar[int] = 1
+
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict, version: int):
+ if version > 1:
+ raise TypeError(f"version > {MyCustom.version}")
+ return MyCustom(data["x"])
Review Comment:
Are you open to "Tabbed" examples yet? :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] BasPH commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
BasPH commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025180942
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,104 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Arguments
+--------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as arguments
+need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``. The following example shows the use of
+a ``Dataset``, which is ``@attr.define`` decorated, together with TaskFlow.
+
+::
+
+ Note: An additional benefit of using ``Dataset`` is that it automatically registers as an ``inlet`` in case it is used as an input argument. It also auto registers as an ``outlet`` if the return value of your task is a ``dataset`` or a ``list[Dataset]]``.
+
+.. code-block:: python
+
+ import json
+ import pendulum
+ import requests
+
+ from airflow import Dataset
+ from airflow.decorators import dag, task
+
+ SRC = Dataset("https://www.ncei.noaa.gov/access/monitoring/climate-at-a-glance/global/time-series/globe/land_ocean/ytd/12/1880-2022.json")
+ now = pendulum.now()
+
+
+ @dag(start_date=now, schedule="daily", catchup=False)
Review Comment:
```suggestion
@dag(start_date=now, schedule="@daily", catchup=False)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025091359
##########
airflow/utils/json.py:
##########
@@ -123,3 +138,113 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder serializes any object that has attr, dataclass or a custom serializer."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ dct = {
+ CLASSNAME: o.__module__ + "." + o.__class__.__qualname__,
+ VERSION: getattr(o.__class__, "version", DEFAULT_VERSION),
+ }
+
+ if hasattr(o, "serialize"):
+ dct[DATA] = getattr(o, "serialize")()
+ return dct
+ elif dataclasses.is_dataclass(o.__class__):
+ data = dataclasses.asdict(o)
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ elif attr.has(o.__class__):
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ if not kwargs.get("object_hook"):
+ kwargs["object_hook"] = self.object_hook
+
+ super().__init__(*args, **kwargs)
+
+ @staticmethod
+ def object_hook(dct: dict) -> object:
+ dct = XComDecoder._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA], version)
+
+ if VERSION in dct and int(dct[VERSION]) > version:
Review Comment:
we are getting this from XCom which has no type safety guarantees. So i'd like to fail here explicitely on the conversion and have a meaningful error (instead of str vs int)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1023900156
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
Review Comment:
Verified: it doesn't work for just `o` - tests do not pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025069885
##########
airflow/decorators/base.py:
##########
@@ -207,17 +209,29 @@ def __init__(
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
def execute(self, context: Context):
+ # todo make this more generic (move to prepare_lineage) so it deals with non taskflow operators
+ # as well
+ for arg in chain(self.op_args, self.op_kwargs.values()):
+ if isinstance(arg, Dataset):
+ self.inlets.append(arg)
return_value = super().execute(context)
return self._handle_output(return_value=return_value, context=context, xcom_push=self.xcom_push)
def _handle_output(self, return_value: Any, context: Context, xcom_push: Callable):
"""
Handles logic for whether a decorator needs to push a single return value or multiple return values.
+ It sets outlets if any datasets are found in the returned value(s)
Review Comment:
Doc style nit. The first paragraph appears in the summary so there should be a blank line. Doesn't matter here as this is a private method so wouldn't appear in docs anyway, but a good habit to get in to.
```suggestion
It sets outlets if any datasets are found in the returned value(s)
```
##########
airflow/models/xcom.py:
##########
@@ -620,32 +621,41 @@ def serialize_value(
if conf.getboolean("core", "enable_xcom_pickling"):
return pickle.dumps(value)
try:
- return json.dumps(value).encode("UTF-8")
- except (ValueError, TypeError):
+ return json.dumps(value, cls=XComEncoder).encode("UTF-8")
+ except (ValueError, TypeError) as ex:
log.error(
- "Could not serialize the XCom value into JSON."
+ f"{ex}."
Review Comment:
```suggestion
"%s."
```
##########
airflow/models/xcom.py:
##########
@@ -620,32 +621,41 @@ def serialize_value(
if conf.getboolean("core", "enable_xcom_pickling"):
return pickle.dumps(value)
try:
- return json.dumps(value).encode("UTF-8")
- except (ValueError, TypeError):
+ return json.dumps(value, cls=XComEncoder).encode("UTF-8")
+ except (ValueError, TypeError) as ex:
log.error(
- "Could not serialize the XCom value into JSON."
+ f"{ex}."
" If you are using pickle instead of JSON for XCom,"
" then you need to enable pickle support for XCom"
- " in your airflow config."
+ " in your airflow config or make sure to decorate your"
+ " object with attr."
Review Comment:
```suggestion
" object with attr.",
ex
```
##########
airflow/utils/json.py:
##########
@@ -123,3 +138,113 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder serializes any object that has attr, dataclass or a custom serializer."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ dct = {
+ CLASSNAME: o.__module__ + "." + o.__class__.__qualname__,
+ VERSION: getattr(o.__class__, "version", DEFAULT_VERSION),
+ }
+
+ if hasattr(o, "serialize"):
+ dct[DATA] = getattr(o, "serialize")()
+ return dct
+ elif dataclasses.is_dataclass(o.__class__):
+ data = dataclasses.asdict(o)
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ elif attr.has(o.__class__):
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ if not kwargs.get("object_hook"):
+ kwargs["object_hook"] = self.object_hook
+
+ super().__init__(*args, **kwargs)
+
+ @staticmethod
+ def object_hook(dct: dict) -> object:
+ dct = XComDecoder._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA], version)
+
+ if VERSION in dct and int(dct[VERSION]) > version:
Review Comment:
We've already checked version is in the dict on L>194, and shouldn't version always be an int? (I guess being "safe" about the type isn't a bad thing though)
```suggestion
if and dct[VERSION] > version:
```
##########
airflow/utils/json.py:
##########
@@ -123,3 +138,113 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder serializes any object that has attr, dataclass or a custom serializer."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ dct = {
+ CLASSNAME: o.__module__ + "." + o.__class__.__qualname__,
+ VERSION: getattr(o.__class__, "version", DEFAULT_VERSION),
+ }
+
+ if hasattr(o, "serialize"):
+ dct[DATA] = getattr(o, "serialize")()
+ return dct
+ elif dataclasses.is_dataclass(o.__class__):
+ data = dataclasses.asdict(o)
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ elif attr.has(o.__class__):
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ if not kwargs.get("object_hook"):
+ kwargs["object_hook"] = self.object_hook
+
+ super().__init__(*args, **kwargs)
+
+ @staticmethod
+ def object_hook(dct: dict) -> object:
+ dct = XComDecoder._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA], version)
Review Comment:
This is the wrong version -- we want to pass the version of the "row", not the current latest version
```suggestion
return getattr(cls, "deserialize")(dct[DATA], dct[VERSION])
```
(But we probably need to do some re-factoring/re-ordering rather than just apply this change)
##########
airflow/utils/json.py:
##########
@@ -40,6 +45,16 @@
log = logging.getLogger(__name__)
+CLASSNAME = "__classname__"
+VERSION = "__version__"
+DATA = "__data__"
+
+OLD_TYPE = "__type"
+OLD_SOURCE = "__source"
+OLD_DATA = "__var"
+
+DEFAULT_VERSION = "0"
Review Comment:
```suggestion
DEFAULT_VERSION = 0
```
I think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1015809492
##########
airflow/decorators/base.py:
##########
@@ -230,7 +230,7 @@ def _handle_output(self, return_value: Any, context: Context, xcom_push: Callabl
if isinstance(return_value, Dataset):
self.outlets.append(return_value)
if isinstance(return_value, list):
- [self.outlets.append(item) for item in return_value if isinstance(item, Dataset)]
+ [self.outlets.append(item) for item in return_value if isinstance(item, Dataset)] # type: ignore[func-returns-value]
Review Comment:
Let’s use a normal for-loop for this. Using a list comprehension in this context is not common in the code base and may cause misunderstandings.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020085216
##########
tests/lineage/test_lineage.py:
##########
@@ -122,10 +130,7 @@ def test_lineage_render(self, dag_maker):
assert op1.inlets[0].url == f1s.format(DEFAULT_DATE)
assert op1.outlets[0].url == f1s.format(DEFAULT_DATE)
- def test_non_attr_outlet(self, dag_maker):
- class A:
- pass
Review Comment:
This test was important to make sure that non-attr managed classes in Inlets/outlets continue to work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020355891
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
Review Comment:
That's not what stackoverflow is saying https://stackoverflow.com/a/38764817 ... but maybe you are right for `serialize` as it is an instance method. `deserialize` is a staticmethod
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1023900489
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
+ }
+ elif attr.has(o.__class__):
+ classname = o.__module__ + "." + o.__class__.__name__
+
+ version = getattr(o, "version", DEFAULT_VERSION)
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: BaseSerialization.serialize(data),
+ }
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(object_hook=self.object_hook, *args, **kwargs)
+
+ def object_hook(self, dct) -> object:
+ dct = self._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if VERSION in dct and int(dct[VERSION]) < version:
+ raise TypeError(
+ "serialized version of %s is newer than module version (%s > %s)",
+ dct[CLASSNAME],
+ dct[VERSION],
+ version,
+ )
+
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA])
Review Comment:
Done. Version checks improved and tested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020083968
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
+ }
+ elif attr.has(o.__class__):
+ classname = o.__module__ + "." + o.__class__.__name__
+
+ version = getattr(o, "version", DEFAULT_VERSION)
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: BaseSerialization.serialize(data),
+ }
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(object_hook=self.object_hook, *args, **kwargs)
+
+ def object_hook(self, dct) -> object:
+ dct = self._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
Review Comment:
```suggestion
version = getattr(cls, "version", DEFAULT_VERSION)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020162266
##########
airflow/utils/json.py:
##########
@@ -40,6 +43,16 @@
log = logging.getLogger(__name__)
+CLASSNAME = "__classname__"
+VERSION = "__version__"
+DATA = "__data__"
Review Comment:
as mentioned `__type` is overloaded and as the deserializer receives the nested dict first it is hard to distinguish between actual type and encoding type. Futhermore, I do think __var and __type are misnomers `__type` is not pythonic and `__var` is incorrect imho (it is the data of an object, not the a variable).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025185101
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -1,19 +1,19 @@
.. Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
.. http://www.apache.org/licenses/LICENSE-2.0
.. Unless required by applicable law or agreed to in writing,
- software distributed under the License is distributed on an
- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- KIND, either express or implied. See the License for the
- specific language governing permissions and limitations
- under the License.
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
Review Comment:
```suggestion
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
```
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -1,19 +1,19 @@
.. Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
Review Comment:
```suggestion
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] marclamberti commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
marclamberti commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1328740750
Love that update! Btw, I tried the DAG given in the example, but the dataset
`SRC = Dataset("https://www.ncei.noaa.gov/access/monitoring/climate-at-a-glance/global/time-series/globe/land_ocean/ytd/12/1880-2022.json")`
doesn't show up in the Dataset view. Is that expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1328793528
@marclamberti yes for now, this is expected. The Dataset view will only reflect "staticly" defined Dataset deps. (I.e. you probably don't talk about this feature much yet.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1310074749
I'm also worried about this changing how existing XCom JSON objects are deserialized -- i.e. before something was just treated as a plain JSON object, but now if it happens to have `__type` key in it then deserialization would fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] BasPH commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
BasPH commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025110283
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
+
+.. code-block:: python
+
+ from typing import ClassVar
+
+
+ class MyCustom:
+ version: ClassVar[int] = 1
+
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict, version: int):
+ if version > 1:
+ raise TypeError(f"version > {MyCustom.version}")
+ return MyCustom(data["x"])
Review Comment:
Referring to https://github.com/apache/airflow/pull/26486 :-) ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025073897
##########
airflow/utils/json.py:
##########
@@ -123,3 +138,113 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder serializes any object that has attr, dataclass or a custom serializer."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ dct = {
+ CLASSNAME: o.__module__ + "." + o.__class__.__qualname__,
+ VERSION: getattr(o.__class__, "version", DEFAULT_VERSION),
+ }
+
+ if hasattr(o, "serialize"):
+ dct[DATA] = getattr(o, "serialize")()
+ return dct
+ elif dataclasses.is_dataclass(o.__class__):
+ data = dataclasses.asdict(o)
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ elif attr.has(o.__class__):
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ if not kwargs.get("object_hook"):
+ kwargs["object_hook"] = self.object_hook
+
+ super().__init__(*args, **kwargs)
+
+ @staticmethod
+ def object_hook(dct: dict) -> object:
+ dct = XComDecoder._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA], version)
+
+ if VERSION in dct and int(dct[VERSION]) > version:
Review Comment:
We've already checked version is in the dict on L>194, and shouldn't version always be an int? (I guess being "safe" about the type isn't a bad thing though)
```suggestion
if dct[VERSION] > version:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1317581553
> Looks good so far, but I've just noticed/realised that this would mean that `orm_deserialize_value` (which currently just calls `deserialize_value`) would import user/dag-author specified code! We'll have to change the impl of orm_deserialize_value so we don't do that.
This is now addressed. PTL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025078487
##########
airflow/models/xcom.py:
##########
@@ -620,32 +621,41 @@ def serialize_value(
if conf.getboolean("core", "enable_xcom_pickling"):
return pickle.dumps(value)
try:
- return json.dumps(value).encode("UTF-8")
- except (ValueError, TypeError):
+ return json.dumps(value, cls=XComEncoder).encode("UTF-8")
+ except (ValueError, TypeError) as ex:
log.error(
- "Could not serialize the XCom value into JSON."
+ f"{ex}."
Review Comment:
That seems to go against the current style?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1015831469
##########
airflow/decorators/base.py:
##########
@@ -230,7 +230,7 @@ def _handle_output(self, return_value: Any, context: Context, xcom_push: Callabl
if isinstance(return_value, Dataset):
self.outlets.append(return_value)
if isinstance(return_value, list):
- [self.outlets.append(item) for item in return_value if isinstance(item, Dataset)]
+ [self.outlets.append(item) for item in return_value if isinstance(item, Dataset)] # type: ignore[func-returns-value]
Review Comment:
Okay I agree that it is not consistent with the rest of the code in this file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1308774921
Do I need to update the json schema for this? There seems to be a reference in the serialization, unfortunately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1017194249
##########
airflow/decorators/base.py:
##########
@@ -210,17 +212,27 @@ def __init__(
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
def execute(self, context: Context):
+ for arg in chain(self.op_args, self.op_kwargs.values()):
+ if isinstance(arg, Dataset):
+ self.inlets.append(arg)
Review Comment:
Shouldn't this be done in the prepare_lineage function (which I think will already called by `super().execute(context)` a few lines down.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1018995338
##########
airflow/decorators/base.py:
##########
@@ -210,17 +212,27 @@ def __init__(
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
def execute(self, context: Context):
+ for arg in chain(self.op_args, self.op_kwargs.values()):
+ if isinstance(arg, Dataset):
+ self.inlets.append(arg)
Review Comment:
I don't think taskflow should be treated separately. It was my starting point (i.e. let taskflow support datasets). I can have a look at what it takes to move it a level higher.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020084477
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
+ }
+ elif attr.has(o.__class__):
+ classname = o.__module__ + "." + o.__class__.__name__
+
+ version = getattr(o, "version", DEFAULT_VERSION)
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: BaseSerialization.serialize(data),
+ }
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ super().__init__(object_hook=self.object_hook, *args, **kwargs)
+
+ def object_hook(self, dct) -> object:
+ dct = self._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if VERSION in dct and int(dct[VERSION]) < version:
+ raise TypeError(
+ "serialized version of %s is newer than module version (%s > %s)",
+ dct[CLASSNAME],
+ dct[VERSION],
+ version,
+ )
+
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA])
Review Comment:
We should probably pass version here too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020078064
##########
airflow/utils/json.py:
##########
@@ -123,3 +136,89 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder allows serializes any object that has attr."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ if hasattr(o, "serialize"):
+ classname = o.__module__ + "." + o.__class__.__name__
+ version = getattr(o.__class__, "version", DEFAULT_VERSION)
+ return {
+ CLASSNAME: classname,
+ VERSION: version,
+ DATA: getattr(o.__class__, "serialize")(o)
Review Comment:
Shouldn't this be
```suggestion
DATA: getattr(o, "serialize")()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020162266
##########
airflow/utils/json.py:
##########
@@ -40,6 +43,16 @@
log = logging.getLogger(__name__)
+CLASSNAME = "__classname__"
+VERSION = "__version__"
+DATA = "__data__"
Review Comment:
as mentioned `__type` is overloaded and as the deserializer receives the nested dict first it is hard to distinguish between actual type and encoding type, which is a builtin. Futhermore, I do think __var and __type are misnomers `__type` is not pythonic and `__var` is incorrect imho (it is the data of an object, not the a variable).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1310151712
> I'm also worried about this changing how existing XCom JSON objects are deserialized -- i.e. before something was just treated as a plain JSON object, but now if it happens to have `__type` key in it then deserialization would fail.
Small correction: if it has '__classname' as a key deserialization would fail.
True, but how big is that risk? If would only happen for xcom values that are custom created or a dictionary that contains '__classname'. What I can do is raise an AttributeError if one tries to serialize with a '__classname' key but the input is a non attr decorated class. I'll add an additional check against `__version` in the deserializer. Chances that a dict would have `__classname` and `__version` specified and is not derserializable are really slim.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1307942871
> '__source' is omitted, it seemed not to be used (could be wrong)
Yes, agree with that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1318641051
LGTM. @BasPH any final comments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025174839
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
Review Comment:
I *think* you are reformatting, but don't know exactly. Please reapply if you don't mind.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025093529
##########
airflow/utils/json.py:
##########
@@ -123,3 +138,113 @@ def dumps(self, obj, **kwargs):
def loads(self, s: str | bytes, **kwargs):
return json.loads(s, **kwargs)
+
+
+# for now separate as AirflowJsonEncoder is non-standard
+class XComEncoder(json.JSONEncoder):
+ """This encoder serializes any object that has attr, dataclass or a custom serializer."""
+
+ def default(self, o: object) -> dict:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ dct = {
+ CLASSNAME: o.__module__ + "." + o.__class__.__qualname__,
+ VERSION: getattr(o.__class__, "version", DEFAULT_VERSION),
+ }
+
+ if hasattr(o, "serialize"):
+ dct[DATA] = getattr(o, "serialize")()
+ return dct
+ elif dataclasses.is_dataclass(o.__class__):
+ data = dataclasses.asdict(o)
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ elif attr.has(o.__class__):
+ # Only include attributes which we can pass back to the classes constructor
+ data = attr.asdict(o, recurse=True, filter=lambda a, v: a.init) # type: ignore[arg-type]
+ dct[DATA] = BaseSerialization.serialize(data)
+ return dct
+ else:
+ return super().default(o)
+
+ def encode(self, o: Any) -> str:
+ if isinstance(o, dict) and CLASSNAME in o:
+ raise AttributeError(f"reserved key {CLASSNAME} found in dict to serialize")
+
+ return super().encode(o)
+
+
+class XComDecoder(json.JSONDecoder):
+ """
+ This decoder deserializes dicts to objects if they contain
+ the `__classname__` key otherwise it will return the dict
+ as is.
+ """
+
+ def __init__(self, *args, **kwargs) -> None:
+ if not kwargs.get("object_hook"):
+ kwargs["object_hook"] = self.object_hook
+
+ super().__init__(*args, **kwargs)
+
+ @staticmethod
+ def object_hook(dct: dict) -> object:
+ dct = XComDecoder._convert(dct)
+
+ if CLASSNAME in dct and VERSION in dct:
+ from airflow.serialization.serialized_objects import BaseSerialization
+
+ cls = import_string(dct[CLASSNAME])
+
+ version = getattr(cls, "version", 0)
+ if hasattr(cls, "deserialize"):
+ return getattr(cls, "deserialize")(dct[DATA], version)
Review Comment:
Good catch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1015827097
##########
airflow/decorators/base.py:
##########
@@ -230,7 +230,7 @@ def _handle_output(self, return_value: Any, context: Context, xcom_push: Callabl
if isinstance(return_value, Dataset):
self.outlets.append(return_value)
if isinstance(return_value, list):
- [self.outlets.append(item) for item in return_value if isinstance(item, Dataset)]
+ [self.outlets.append(item) for item in return_value if isinstance(item, Dataset)] # type: ignore[func-returns-value]
Review Comment:
I think the list comprehension is more clear than a normal for loop and the for loop is needlessly complicated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1308697500
> Need to think about how this might affect loading from xcom existing values? Have you already considered that?
Yes I did. In general the function is backwards compatible. However, if previously something was serialized with `__classname` as part of the resulting json, it would be affected. I estimated the chance of that happening being close to zero. I chose not to return the dict in case of failure to deserialize in case of `__classname` being present as it would result in unpredicatable behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1017893141
##########
airflow/decorators/base.py:
##########
@@ -210,17 +212,27 @@ def __init__(
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
def execute(self, context: Context):
+ for arg in chain(self.op_args, self.op_kwargs.values()):
+ if isinstance(arg, Dataset):
+ self.inlets.append(arg)
Review Comment:
Are you suggesting that we would make this more generic? It now only affects the `DecoratedOperator` moving it to prepare_lineage (part of `pre_execute`) would make it affect all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1018919273
##########
airflow/decorators/base.py:
##########
@@ -210,17 +212,27 @@ def __init__(
super().__init__(task_id=task_id, **kwargs_to_upstream, **kwargs)
def execute(self, context: Context):
+ for arg in chain(self.op_args, self.op_kwargs.values()):
+ if isinstance(arg, Dataset):
+ self.inlets.append(arg)
Review Comment:
That was my thought (specifically why should taskflow be treated specially?) but I haven't thought that deeply to consider what the implications might be.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #27540:
URL: https://github.com/apache/airflow/pull/27540#issuecomment-1310165850
> Chances that a dict would have __classname and __version specified and is not derserializable are really slim.
If you do that, and add a "significant" note so it goes in the changelog (see end of PR template for short instructions) then sounds good to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1020358570
##########
tests/lineage/test_lineage.py:
##########
@@ -122,10 +130,7 @@ def test_lineage_render(self, dag_maker):
assert op1.inlets[0].url == f1s.format(DEFAULT_DATE)
assert op1.outlets[0].url == f1s.format(DEFAULT_DATE)
- def test_non_attr_outlet(self, dag_maker):
- class A:
- pass
Review Comment:
Non attr (or actually non serializable) inlets/outlets didn't work, they were just ignored. I don't think that is correct behavioral integrity. This leads to surprises where you supplied something as an outlet somewhere and made it an inlet to something else and it never 'arrived' an you don't know why.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] BasPH commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
BasPH commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025100952
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
+
+.. code-block:: python
+
+ from typing import ClassVar
+
+
+ class MyCustom:
+ version: ClassVar[int] = 1
+
+ def __init__(self, x):
+ self.x = x
+
+ def serialize(self) -> dict:
+ return dict({"x": self.x})
+
+ @staticmethod
+ def deserialize(data: dict, version: int):
+ if version > 1:
+ raise TypeError(f"version > {MyCustom.version}")
+ return MyCustom(data["x"])
Review Comment:
Could you demonstrate a taskflow task using the `MyCustom` class in the example?
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
Review Comment:
```suggestion
TaskFlow uses XCom to pass variables to each task. This requires variables that are used as arguments
to tasks to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to pass custom objects, or if you want to control serialization yourself. To do so, implement the ``serialize()`` method and ``deserialize(data: dict, version: int)`` staticmethod in your class. Like so:
```
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
Review Comment:
```suggestion
Passing Arbitrary Objects As Arguments
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025100348
##########
airflow/models/xcom.py:
##########
@@ -620,32 +621,41 @@ def serialize_value(
if conf.getboolean("core", "enable_xcom_pickling"):
return pickle.dumps(value)
try:
- return json.dumps(value).encode("UTF-8")
- except (ValueError, TypeError):
+ return json.dumps(value, cls=XComEncoder).encode("UTF-8")
+ except (ValueError, TypeError) as ex:
log.error(
- "Could not serialize the XCom value into JSON."
+ f"{ex}."
Review Comment:
```
- id: check-lazy-logging
name: Check that all logging methods are lazy
entry: ./scripts/ci/pre_commit/pre_commit_check_lazy_logging.py
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025181217
##########
docs/apache-airflow/concepts/taskflow.rst:
##########
@@ -81,6 +81,49 @@ To use logging from your task functions, simply import and use Python's logging
Every logging line created this way will be recorded in the task log.
+Passing Arbitrary Objects As Parameters
+---------------------------------------
+
+.. versionadded:: 2.5.0
+
+As mentioned TaskFlow uses XCom to pass variables to each task. This requires that variables that are used as parameters
+to tasks need to be able to be serialized. Airflow out of the box supports all built-in types (like int or str) and it
+supports objects that are decorated with ``@dataclass`` or ``@attr.define``, but it will need some help if you want to
+pass custom objects or if you want to control serialization yourself. To do so add the ``serialize()`` method to your
+class and the staticmethod ``deserialize(data: dict, version: int)`` to your class. Like so:
Review Comment:
If you can do that again that would be appreciated, I am getting word blind here :-P
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bolkedebruin commented on a diff in pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
bolkedebruin commented on code in PR #27540:
URL: https://github.com/apache/airflow/pull/27540#discussion_r1025108099
##########
airflow/models/xcom.py:
##########
@@ -620,32 +621,41 @@ def serialize_value(
if conf.getboolean("core", "enable_xcom_pickling"):
return pickle.dumps(value)
try:
- return json.dumps(value).encode("UTF-8")
- except (ValueError, TypeError):
+ return json.dumps(value, cls=XComEncoder).encode("UTF-8")
+ except (ValueError, TypeError) as ex:
log.error(
- "Could not serialize the XCom value into JSON."
+ f"{ex}."
Review Comment:
Okido :-P geez :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] BasPH merged pull request #27540: Allow datasets to be used in taskflow
Posted by GitBox <gi...@apache.org>.
BasPH merged PR #27540:
URL: https://github.com/apache/airflow/pull/27540
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org