Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

MLTransform #26795

Merged
merged 68 commits into from
Jul 6, 2023
Merged
Show file tree
Hide file tree
Changes from 27 commits
Commits
Show all changes
68 commits
Select commit Hold shift + click to select a range
6c641ec
Initial work on MLTransform and ProcessHandler
AnandInguva May 10, 2023
9c93dc0
Support for containers: List, Dict[str, np.ndarray]
AnandInguva May 15, 2023
89e889d
Add min, max, artifacts for scale_0_to_1
AnandInguva May 16, 2023
5548103
Add more transform functions and artifacts
AnandInguva May 19, 2023
f5d050a
Add generic type annotations
AnandInguva May 19, 2023
6681916
Add unit tests
AnandInguva May 30, 2023
3be1cfd
Add support for saving intermediate results for a transform
AnandInguva May 30, 2023
6caba7e
Add schema to the output PCollection
AnandInguva May 31, 2023
901a74c
Remove MLTransformOutput and return Row instead with schema
AnandInguva May 31, 2023
361e0bb
Convert primitive type to list using a DoFn. Remove FixedLenFeatureSpec
AnandInguva Jun 1, 2023
681d164
Add append_transform to the ProcessHandler
AnandInguva Jun 1, 2023
eac8b3f
Remove param self.has_artifacts, add artifact_location to handler..an…
AnandInguva Jun 2, 2023
011d5d1
Move tensorflow import into the try except catch
AnandInguva Jun 2, 2023
def7eb4
Add type annotations for the data transforms
AnandInguva Jun 2, 2023
1a0a0ed
Add tft test in tox.ini
AnandInguva Jun 5, 2023
2393254
Add step name for TFTProcessHandler
AnandInguva Jun 5, 2023
f25618e
Remove unsupported tft versions
AnandInguva Jun 5, 2023
4256c99
Fix mypy
AnandInguva Jun 5, 2023
df73361
Refactor TFTProcessHandlerDict to TFTProcessHandlerSchema
AnandInguva Jun 5, 2023
4497bb5
Update doc for data processing transforms
AnandInguva Jun 6, 2023
77b3634
Fix checking the typing container types
AnandInguva Jun 6, 2023
baf1ae7
Refactor code
AnandInguva Jun 6, 2023
044f509
Fail TFTProcessHandler on a non-global window PColl
AnandInguva Jun 6, 2023
c312aef
Remove underscore
AnandInguva Jun 7, 2023
68a2529
Remove high level functions
AnandInguva Jun 7, 2023
e6ef468
Add TFIDF
AnandInguva Jun 9, 2023
2be4ba6
Fix tests with new changes[WIP]
AnandInguva Jun 9, 2023
7a290e2
Fix tests
AnandInguva Jun 10, 2023
c2a1fae
Refactor class name to CamelCase and remove kwrags
AnandInguva Jun 10, 2023
21dadb1
use is_default instead of isinstance
AnandInguva Jun 16, 2023
df05169
Remove falling back to staging location for artifact location
AnandInguva Jun 16, 2023
42fd6c4
Add TFIDF tests
AnandInguva Jun 16, 2023
5c6dcb4
Remove __str__
AnandInguva Jun 16, 2023
43d24ad
Refactor skip statement
AnandInguva Jun 16, 2023
618b2fa
Add utils for fetching artifacts on compute and apply vocab
AnandInguva Jun 16, 2023
a814650
Make ProcessHandler internal class
AnandInguva Jun 17, 2023
0a61955
Only run analyze stage when transform_fn(artifacts) is not computed b…
AnandInguva Jun 22, 2023
33f8fb2
Fail if pipeline has non default window during artifact producing stage
AnandInguva Jun 22, 2023
bc22e9f
Add support for Dict, recordbatch and introduce artifact_mode
AnandInguva Jun 23, 2023
4e07f7d
Hide process_handler from user. Make TFTProcessHandler as default
AnandInguva Jun 23, 2023
eeed56c
Refactor few tests
AnandInguva Jun 23, 2023
9eed989
Comment a test
AnandInguva Jun 23, 2023
3453b9f
Save raw_data_meta_data so that it can be used during consume stage
AnandInguva Jun 23, 2023
3e8f198
Refactor code
AnandInguva Jun 23, 2023
e8a3686
Add test on artifacts
AnandInguva Jun 23, 2023
72ea029
Fix imports
AnandInguva Jun 26, 2023
55b04e8
Add tensorflow_metadata to pydocs
AnandInguva Jun 26, 2023
b65ff05
Merge remote-tracking branch 'upstream/master' into mltransform
AnandInguva Jun 26, 2023
00fb944
Fix test
AnandInguva Jun 26, 2023
f11d02b
Add TFIDF to import
AnandInguva Jun 26, 2023
7b2200f
Add basic example
AnandInguva Jun 26, 2023
bca2dda
Remove redundant logging statements
AnandInguva Jun 27, 2023
295a80d
Add test for multiple columns on MLTransform
AnandInguva Jun 29, 2023
1d0b5b1
Add todo about what to do when new process handler is introduced
AnandInguva Jun 29, 2023
64bba5e
Add abstractmethod decorator
AnandInguva Jun 29, 2023
034a066
Edit Error message
AnandInguva Jun 29, 2023
1eef0e7
Update docs, error messages
AnandInguva Jun 29, 2023
4ed94c7
Remove record batch input/output arg
AnandInguva Jun 30, 2023
2e6c5ac
Modify generic types
AnandInguva Jun 30, 2023
bf81d46
Fix import sort
AnandInguva Jun 30, 2023
0860489
Merge remote-tracking branch 'upstream/master' into mltransform
AnandInguva Jun 30, 2023
1dcdaa8
Fix mypy errors - best effort
AnandInguva Jul 5, 2023
bb9336a
Fix tests
AnandInguva Jul 5, 2023
17a4eb1
Add TFTOperation doc
AnandInguva Jul 5, 2023
20f416d
Rename tft_transform to tft
AnandInguva Jul 5, 2023
ba33cb7
Fix hadler_test
AnandInguva Jul 5, 2023
f0c023b
Fix base_test
AnandInguva Jul 5, 2023
a315091
Fix pydocs
AnandInguva Jul 5, 2023
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions sdks/python/apache_beam/ml/transforms/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
#
# 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
#
# 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.
#
116 changes: 116 additions & 0 deletions sdks/python/apache_beam/ml/transforms/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
#
# 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
#
# 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.

from typing import Generic
from typing import TypeVar

import apache_beam as beam

# TODO: Abstract methods are not getting pickled with dill.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this TODO still apply? What are the consequences?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not relevant anymore. I tried today and I wasn't able to reproduce it now

# https://github.com/uqfoundation/dill/issues/332
# import abc

__all__ = ['MLTransform', 'ProcessHandler']

TransformedDatasetT = TypeVar('TransformedDatasetT')
TransformedMetadataT = TypeVar('TransformedMetadataT')

# Input/Output types to the MLTransform.
ExampleT = TypeVar('ExampleT')
MLTransformOutputT = TypeVar('MLTransformOutputT')

# Input to the process data. This could be same or different from ExampleT.
damccorm marked this conversation as resolved.
Show resolved Hide resolved
ProcessInputT = TypeVar('ProcessInputT')
# Output of the process data. This could be same or different
# from MLTransformOutputT
ProcessOutputT = TypeVar('ProcessOutputT')

# Input to the apply() method of BaseOperation.
OperationInputT = TypeVar('OperationInputT')
# Output of the apply() method of BaseOperation.
OperationOutputT = TypeVar('OperationOutputT')


class BaseOperation(Generic[OperationInputT, OperationOutputT]):
def apply(
self, inputs: OperationInputT, column_name: str, *args,
**kwargs) -> OperationOutputT:
"""
Define any processing logic in the apply() method.
processing logics are applied on inputs and returns a transformed
output.
Args:
inputs: input data.
"""
raise NotImplementedError


class ProcessHandler(Generic[ProcessInputT, ProcessOutputT]):
def process_data(
self, pcoll: beam.PCollection[ProcessInputT]
) -> beam.PCollection[ProcessOutputT]:
"""
Logic to process the data. This will be the entrypoint in
beam.MLTransform to process incoming data.
"""
raise NotImplementedError

def append_transform(self, transform: BaseOperation):
raise NotImplementedError


class MLTransform(beam.PTransform[beam.PCollection[ExampleT],
beam.PCollection[MLTransformOutputT]],
Generic[ExampleT, MLTransformOutputT, ]):
AnandInguva marked this conversation as resolved.
Show resolved Hide resolved
def __init__(
self,
process_handler: ProcessHandler[ExampleT, MLTransformOutputT],
):
"""
Args:
process_handler: A ProcessHandler instance that defines the logic to
process the data.
"""
self._process_handler = process_handler

def expand(
self, pcoll: beam.PCollection[ExampleT]
) -> beam.PCollection[MLTransformOutputT]:
"""
This is the entrypoint for the MLTransform. This method will
invoke the process_data() method of the ProcessHandler instance
to process the incoming data.

process_data takes in a PCollection and applies the PTransforms
necessary to process the data and returns a PCollection of
transformed data.
Args:
pcoll: A PCollection of ExampleT type.
Returns:
A PCollection of MLTransformOutputT type.
"""
return self._process_handler.process_data(pcoll)

def with_transform(self, transform: BaseOperation):
"""
Add a transform to the MLTransform pipeline.
Args:
transform: A BaseOperation instance.
Returns:
A MLTransform instance.
"""
self._process_handler.append_transform(transform)
return self
Loading