[BE] Unify Map style DataPipes to support input_col and output_col syntax #562

VitalyFedyunin · 2022-06-29T20:23:37Z

Stack from ghstack (oldest at bottom):

Differential Revision: D40146676

…ntax [ghstack-poisoned]

…ntax ghstack-source-id: ac06eb69a0dd654ae6ef6b9a0086cb5625ac487b Pull Request resolved: #562

NivekT

Should we also add one for flatmap?

NivekT · 2022-06-29T21:09:47Z

torchdata/datapipes/iter/util/common.py

+
+
+class MapTemplateIterDataPipe(Mapper):
+ def __init__(self, source_datapipe, input_col=None, output_col=None) -> None:


Let's add a docstring for the purpose of this class.

Add the class MapTemplate to RST documnetation file as well.

I will update DP contributors doc and add docstrings

VitalyFedyunin · 2022-06-29T21:18:23Z

Surely doing it for flatmap now, plus will attempt to convert several inner DataPipes to this approach.

ejguan

I think it's better to move this class to PyTorch Core for Mapper and Filter.

…tput_col syntax" [ghstack-poisoned]

…ntax ghstack-source-id: bc08ee551c3b1b72158dadd5701d1d1b15b496e5 Pull Request resolved: #562

VitalyFedyunin · 2022-06-30T20:58:13Z

UPD: Temporary make a copy of FlatMapper
UPD: Rewrote JsonParser as example.

VitalyFedyunin · 2022-07-06T20:00:04Z

I think it's better to move this class to PyTorch Core for Mapper and Filter.

To be honest I prefer to keep as much as possible inside this repo. This will make future deprecation easier.

…tput_col syntax" [ghstack-poisoned]

…ntax ghstack-source-id: 9e62d23ba6008a32f5b71e32da93896571d598d0 Pull Request resolved: #562

…tput_col syntax" [ghstack-poisoned]

…ntax ghstack-source-id: 3227ba7b13a3477fb294708fb0bb551399d70747 Pull Request resolved: #562

…tput_col syntax" [ghstack-poisoned]

…ntax ghstack-source-id: e7320cd563b122d952dbf0dfacb4fe6d39804b51 Pull Request resolved: #562

VitalyFedyunin · 2022-07-08T15:57:53Z

test/test_local_io.py

+ return {"file_name": data[0], "stream": data[1]}
+
+ dict_dp = datapipe_nonempty.map(as_dict)
+ parsed_dp = dict_dp.parse_json_files(input_col="stream", output_col="json")


This is the most important part of the PR. IMO it will simplify some of the domains code.

ejguan · 2022-07-08T17:18:22Z

torchdata/datapipes/iter/util/common.py

+@functional_datapipe("flatmap_proto")
+class FlatMapperProtoIterDataPipe(IterDataPipe[T_co]):


Do you want to directly change the FlatMapper here?

data/torchdata/datapipes/iter/transform/callable.py

Lines 93 to 94 in 14b2291

@functional_datapipe("flatmap")

class FlatMapperIterDataPipe(IterDataPipe[T_co]):

Yes I do, was making sure I cover all use cases before replacement.

ejguan · 2022-07-08T17:20:04Z

torchdata/datapipes/iter/util/jsonparser.py

 self.kwargs = kwargs

- def __iter__(self) -> Iterator[Tuple[str, Dict]]:
- for file_name, stream in self.source_datapipe:
+ def _map(self, stream: IO):


Do you think we should change the type to StreamWrapper here?

We are fine with any stream as input.

ejguan · 2022-07-08T21:04:01Z

torchdata/datapipes/iter/util/common.py

+ def _map(self, *args, **kwargs) -> T_co:
+ raise NotImplementedError


BTW, we might want to use abc. abstractmethod here to guarantee to implement the method at the construct time.

ejguan · 2022-07-08T21:04:08Z

torchdata/datapipes/iter/util/common.py

+ def _flatmap(self, *args, **kwargs) -> Iterable[T_co]:
+ raise NotImplementedError


…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin · 2022-10-05T19:31:17Z

Direct use of abstract leads to:

torchdata/datapipes/iter/util/jsonparser.py:40:5: error: Signature of "_map"
incompatible with supertype "MapTemplateIterDataPipe"  [override]
        def _map(self, stream: IO):

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin · 2022-10-06T15:58:03Z

@VitalyFedyunin has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

ejguan · 2022-10-06T18:04:18Z

torchdata/datapipes/iter/transform/callable.py

+ _check_unpickable_fn(fn)
+ self.fn = fn # type: ignore[assignment]
+
+ self.input_col = input_col


We can also validate input_col with fn using torch.utils.data.datapipes.utils.common.validate_input_col
https://github.com/pytorch/pytorch/blob/cfb9d0d23314fd28be118b6ca280ded55364e71c/torch/utils/data/datapipes/utils/common.py#L24

NivekT

nit: This PR is fine (with nit comments) but there is something unsatisfying about how input_col and output_col work in the case where I want to select an element from a tuple/list?

Note that we do not have to address the below within this PR.

A common return pattern that we have is (Path, Stream) from archives/streams/etc. Let's say I opened a file and now I have a DataPipe of tuples of (Path, Stream), and I want to throw away the path and parse the stream. I will have to do something like this

datapipe = FileOpener(json_files, mode="b")
datapipe = datapipe.parse_json_files()
datapipe = datapipe.map(lambda x: x[1])  # Ideally this should not be needed, and handled by `input_col`, `output_col`

I feel like ideally the input_col and output_col argument should allow for selection (unpack when the elements are list of length 1 or tuple of length 1).

We likely also want to allow the case where the input is not a list/tuple:

datapipe = IterableWrapper([stream1, stream2])
datapipe = datapipe.parse_json_files()  # I think we should support this, but it currently doesn't work

What do you two think?

cc: @ejguan

NivekT · 2022-10-06T18:37:42Z

torchdata/datapipes/iter/util/common.py

+class MapTemplateIterDataPipe(Mapper[T_co]):
+ def __init__(self, source_datapipe, input_col=None, output_col=None) -> None:
+ fn = getattr(self, "_map")
+ assert fn is not None


Suggested change

assert fn is not None

assert fn is not None, "`MapTemplate` expects subclass to have attribute `_map` defined, but it is `None`."

NivekT · 2022-10-06T19:40:49Z

torchdata/datapipes/iter/util/jsonparser.py

 [('1.json', ['foo', {'bar': ['baz', None, 1.0, 2]}]), ('2.json', {'__complex__': True, 'real': 1, 'imag': 2})]
 """


Adding usage example for input_col argument

Suggested change

[('1.json', ['foo', {'bar': ['baz', None, 1.0, 2]}]), ('2.json', {'__complex__': True, 'real': 1, 'imag': 2})]

"""

[('1.json', ['foo', {'bar': ['baz', None, 1.0, 2]}]), ('2.json', {'__complex__': True, 'real': 1, 'imag': 2})]

>>> datapipe_extra_col = datapipe1.map(lambda x: (x[0], "extra", x[1]))

>>> # Use `input_col` if the stream is not in position 1

>>> json_dp2 = datapipe_extra_col.parse_json_files(input_col=2)

>>> list(json_dp2)

[

("1.json", "extra", ["foo", {"bar": ["baz", None, 1.0, 2]}]),

("2.json", "extra", {"__complex__": True, "real": 1, "imag": 2}),

]

"""

NivekT · 2022-10-06T19:41:47Z

torchdata/datapipes/iter/util/jsonparser.py



 @functional_datapipe("parse_json_files")
-class JsonParserIterDataPipe(IterDataPipe[Tuple[str, Dict]]):
+class JsonParserIterDataPipe(MapTemplate):
 r"""
 Reads from JSON data streams and yields a tuple of file name and JSON data (functional name: ``parse_json_files``).



Need to update docstring to add new arguments

NivekT · 2022-10-06T20:01:04Z

torchdata/datapipes/iter/transform/callable.py

@@ -162,6 +165,115 @@ def __len__(self) -> int:
 raise TypeError(f"{type(self).__name__}'s length relies on the output of its function.")


+# TODO(VitalyFedyunin): Replacing FlatMapperIterDataPipe will require BC breaking change of input_col behaviour


Is this because when input_col is specified`, the new version will try to return a list/tuple if the original input is a list/tuple?

ejguan · 2022-10-06T20:03:20Z

nit: This PR is fine (with nit comments) but there is something unsatisfying about how input_col and output_col work in the case where I want to select an element from a tuple/list?

Note that we do not have to address the below within this PR.

A common return pattern that we have is (Path, Stream) from archives/streams/etc. Let's say I opened a file and now I have a DataPipe of tuples of (Path, Stream), and I want to throw away the path and parse the stream. I will have to do something like this
datapipe = FileOpener(json_files, mode="b")
datapipe = datapipe.parse_json_files()
datapipe = datapipe.map(lambda x: x[1])  # Ideally this should not be needed, and handled by `input_col`, `output_col`
I feel like ideally the input_col and output_col argument should allow for selection (unpack when the elements are list of length 1 or tuple of length 1).

We likely also want to allow the case where the input is not a list/tuple:
datapipe = IterableWrapper([stream1, stream2])
datapipe = datapipe.parse_json_files()  # I think we should support this, but it currently doesn't work
What do you two think?

cc: @ejguan

I think we have discussed about it earlier about the expected behavior of the combination of input_col and output_col. Currently, when input_col is specified buy output_col is not, we will do in-place operation. In order to support selection, i can't find a good value for output_col to indicate that we want to select a single column.

datapipe = datapipe.parse_json_files()  # I think we should support this, but it currently doesn't work

Do you mean you want it still works with stream input rather than a tuple of (pathname, stream)

NivekT · 2022-10-06T20:19:34Z

I think we have discussed about it earlier about the expected behavior of the combination of input_col and output_col. Currently, when input_col is specified buy output_col is not, we will do in-place operation. In order to support selection, i can't find a good value for output_col to indicate that we want to select a single column.

One potential solution is to have output_col=[] or output_col=tuple() as selection. There may be side effects that I currently cannot think of.

Do you mean you want it still works with stream input rather than a tuple of (pathname, stream)

Yes, it may require some adjustment to how input_col and MapTemplate works to handle non-list/tuple inputs.

NivekT · 2022-10-06T20:23:36Z

Sorry I accidentally edited your comment while quoting it.

ejguan · 2022-10-06T20:33:46Z

Yes, it may require some adjustment to how input_col and MapTemplate works to handle non-list/tuple inputs.

I think it's doable after this PR. We can let JsonParser be a MapTemplate with input_col=1 by default to prevent BC breaking.
Users can specify input_col=None to make it directly working with stream.

facebook-github-bot · 2022-12-02T16:05:41Z

Hi @VitalyFedyunin!

Thank you for your pull request.

We require contributors to sign our Contributor License Agreement, and yours needs attention.

You currently have a record in our system, but the CLA is no longer valid, and will need to be resubmitted.

Process

In order for us to review and merge your suggested changes, please sign at https://code.facebook.com/cla. If you are contributing on behalf of someone else (eg your employer), the individual CLA may not be sufficient and your employer may need to sign the corporate CLA.

Once the CLA is signed, our tooling will perform checks and validations. Afterwards, the pull request will be tagged with CLA signed. The tagging process may take up to 1 hour after signing. Please give it that time before contacting us about it.

If you have received this in error or have any questions, please contact us at [email protected]. Thanks!

[BE] Unify Map style DataPipes to support input_col and output_col sy…

45e648e

…ntax [ghstack-poisoned]

VitalyFedyunin added a commit that referenced this pull request Jun 29, 2022

[BE] Unify Map style DataPipes to support input_col and output_col sy…

6f96bd3

…ntax ghstack-source-id: ac06eb69a0dd654ae6ef6b9a0086cb5625ac487b Pull Request resolved: #562

facebook-github-bot added the CLA Signed This label is managed by the Facebook bot. Authors need to sign the CLA before a PR can be reviewed. label Jun 29, 2022

VitalyFedyunin requested review from NivekT and ejguan June 29, 2022 20:24

NivekT reviewed Jun 29, 2022

View reviewed changes

ejguan reviewed Jun 29, 2022

View reviewed changes

VitalyFedyunin mentioned this pull request Jun 30, 2022

WIP FlatMap #565

Closed

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

23ff4d7

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin added a commit that referenced this pull request Jun 30, 2022

[BE] Unify Map style DataPipes to support input_col and output_col sy…

426f318

…ntax ghstack-source-id: bc08ee551c3b1b72158dadd5701d1d1b15b496e5 Pull Request resolved: #562

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

1e360df

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin added a commit that referenced this pull request Jul 6, 2022

[BE] Unify Map style DataPipes to support input_col and output_col sy…

78ca51c

…ntax ghstack-source-id: 9e62d23ba6008a32f5b71e32da93896571d598d0 Pull Request resolved: #562

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

729e1c4

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin added a commit that referenced this pull request Jul 6, 2022

[BE] Unify Map style DataPipes to support input_col and output_col sy…

d0ac37c

…ntax ghstack-source-id: 3227ba7b13a3477fb294708fb0bb551399d70747 Pull Request resolved: #562

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

5926327

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin added a commit that referenced this pull request Jul 6, 2022

[BE] Unify Map style DataPipes to support input_col and output_col sy…

07e9ed3

…ntax ghstack-source-id: e7320cd563b122d952dbf0dfacb4fe6d39804b51 Pull Request resolved: #562

VitalyFedyunin commented Jul 8, 2022

View reviewed changes

ejguan reviewed Jul 8, 2022

View reviewed changes

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

03c8562

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin mentioned this pull request Jul 13, 2022

Changing 1-to-M behaviour of on_disk_cache. #652

Closed

VitalyFedyunin added 4 commits October 5, 2022 14:26

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

e650d59

…tput_col syntax" [ghstack-poisoned]

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

b041059

…tput_col syntax" [ghstack-poisoned]

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

f603e40

…tput_col syntax" [ghstack-poisoned]

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

85c082c

…tput_col syntax" [ghstack-poisoned]

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

3f24e0a

…tput_col syntax" [ghstack-poisoned]

Update on "[BE] Unify Map style DataPipes to support input_col and ou…

ed4a966

…tput_col syntax" [ghstack-poisoned]

VitalyFedyunin requested review from ejguan and NivekT October 5, 2022 22:26

VitalyFedyunin mentioned this pull request Oct 5, 2022

[WIP] Add debug lines for on_disk_cache #809

Open

ejguan approved these changes Oct 6, 2022

View reviewed changes

NivekT approved these changes Oct 6, 2022

View reviewed changes

NivekT reviewed Oct 6, 2022

View reviewed changes

SvenDS9 mentioned this pull request Mar 2, 2023

Using datapipes on only parts of the input #1030

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BE] Unify Map style DataPipes to support input_col and output_col syntax #562

[BE] Unify Map style DataPipes to support input_col and output_col syntax #562

VitalyFedyunin commented Jun 29, 2022 •

edited

NivekT left a comment

NivekT Jun 29, 2022

VitalyFedyunin Jul 8, 2022

VitalyFedyunin commented Jun 29, 2022

ejguan left a comment

VitalyFedyunin commented Jun 30, 2022

VitalyFedyunin commented Jul 6, 2022

VitalyFedyunin Jul 8, 2022

ejguan Jul 8, 2022

VitalyFedyunin Jul 8, 2022

ejguan Jul 8, 2022

VitalyFedyunin Jul 8, 2022

ejguan Jul 8, 2022

ejguan Jul 8, 2022

VitalyFedyunin commented Oct 5, 2022

VitalyFedyunin commented Oct 6, 2022

ejguan Oct 6, 2022

NivekT left a comment •

edited

NivekT Oct 6, 2022

NivekT Oct 6, 2022

NivekT Oct 6, 2022

NivekT Oct 6, 2022

ejguan commented Oct 6, 2022 •

edited by NivekT

NivekT commented Oct 6, 2022

NivekT commented Oct 6, 2022

ejguan commented Oct 6, 2022

facebook-github-bot commented Dec 2, 2022



		class MapTemplateIterDataPipe(Mapper):
		def __init__(self, source_datapipe, input_col=None, output_col=None) -> None:

		@functional_datapipe("flatmap_proto")
		class FlatMapperProtoIterDataPipe(IterDataPipe[T_co]):

	@functional_datapipe("flatmap")
	class FlatMapperIterDataPipe(IterDataPipe[T_co]):

		def _map(self, args, *kwargs) -> T_co:
		raise NotImplementedError

		def _flatmap(self, args, *kwargs) -> Iterable[T_co]:
		raise NotImplementedError

	assert fn is not None
	assert fn is not None, "`MapTemplate` expects subclass to have attribute `_map` defined, but it is `None`."

		[('1.json', ['foo', {'bar': ['baz', None, 1.0, 2]}]), ('2.json', {'__complex__': True, 'real': 1, 'imag': 2})]
		"""

- [('1.json', ['foo', {'bar': ['baz', None, 1.0, 2]}]), ('2.json', {'__complex__': True, 'real': 1, 'imag': 2})]
- """
+ [('1.json', ['foo', {'bar': ['baz', None, 1.0, 2]}]), ('2.json', {'__complex__': True, 'real': 1, 'imag': 2})]
+ >>> datapipe_extra_col = datapipe1.map(lambda x: (x[0], "extra", x[1]))
+ >>> # Use `input_col` if the stream is not in position 1
+ >>> json_dp2 = datapipe_extra_col.parse_json_files(input_col=2)
+ >>> list(json_dp2)
+ [
+ ("1.json", "extra", ["foo", {"bar": ["baz", None, 1.0, 2]}]),
+ ("2.json", "extra", {"__complex__": True, "real": 1, "imag": 2}),
+ ]
+ """

		@@ -162,6 +165,115 @@ def __len__(self) -> int:
		raise TypeError(f"{type(self).__name__}'s length relies on the output of its function.")


		# TODO(VitalyFedyunin): Replacing FlatMapperIterDataPipe will require BC breaking change of input_col behaviour

[BE] Unify Map style DataPipes to support input_col and output_col syntax #562

Are you sure you want to change the base?

[BE] Unify Map style DataPipes to support input_col and output_col syntax #562

Conversation

VitalyFedyunin commented Jun 29, 2022 • edited

NivekT left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VitalyFedyunin commented Jun 29, 2022

ejguan left a comment

Choose a reason for hiding this comment

VitalyFedyunin commented Jun 30, 2022

VitalyFedyunin commented Jul 6, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VitalyFedyunin commented Oct 5, 2022

VitalyFedyunin commented Oct 6, 2022

Choose a reason for hiding this comment

NivekT left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ejguan commented Oct 6, 2022 • edited by NivekT

NivekT commented Oct 6, 2022

NivekT commented Oct 6, 2022

ejguan commented Oct 6, 2022

facebook-github-bot commented Dec 2, 2022

Process

VitalyFedyunin commented Jun 29, 2022 •

edited

NivekT left a comment •

edited

ejguan commented Oct 6, 2022 •

edited by NivekT