You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/06/01 17:08:21 UTC
[GitHub] [beam] ryanthompson591 opened a new pull request, #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
ryanthompson591 opened a new pull request, #17800:
URL: https://github.com/apache/beam/pull/17800
Scikit learn developers build pipelines using pandas very often and we should support it if we want to support pandas pipelines.
Here's [an example from their site](https://scikit-learn.org/stable/auto_examples/compose/plot_column_transformer_mixed_types.html).
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ryanthompson591 commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
ryanthompson591 commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1146225373
R: @TheNeuralBit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1152571026
I'll go ahead and merge this for the base support, but as discussed offline, let's follow-up with a change to break out the numpy array and dataframe implementations into separate ModelLoaders, and mark the dataframe one experimental
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ryanthompson591 commented on a diff in pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
ryanthompson591 commented on code in PR #17800:
URL: https://github.com/apache/beam/pull/17800#discussion_r892469936
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -71,7 +72,7 @@ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
# vectorize data for better performance
vectorized_batch = pandas.concat(batch, axis=0)
predictions = model.predict(vectorized_batch)
- splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
+ splits = [vectorized_batch.iloc[[i]] for i in range(len(batch))]
Review Comment:
I think for now it is ok to raise an exception.
In the future we should support multiple rows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #17800:
URL: https://github.com/apache/beam/pull/17800#discussion_r892963554
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -41,20 +43,57 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
def run_inference(
- self, batch: List[numpy.ndarray], model: BaseEstimator,
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator,
**kwargs) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
Review Comment:
Could we have two separate InferenceRunner implementations, one for np.array and one for DataFrame? Then we wouldn't need to branch on the type here for every batch. The user could indicate which they prefer with an argument on the model loader.
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -41,20 +43,57 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
def run_inference(
- self, batch: List[numpy.ndarray], model: BaseEstimator,
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator,
**kwargs) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
+ raise ValueError('Unsupported data type.')
+
+ @staticmethod
+ def _predict_np_array(batch: List[numpy.ndarray],
+ model: Any) -> Iterable[PredictionResult]:
# vectorize data for better performance
vectorized_batch = numpy.stack(batch, axis=0)
predictions = model.predict(vectorized_batch)
return [PredictionResult(x, y) for x, y in zip(batch, predictions)]
- def get_num_bytes(self, batch: List[numpy.ndarray]) -> int:
+ @staticmethod
+ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
+ model: Any) -> Iterable[PredictionResult]:
+ # sklearn_inference currently only supports single rowed dataframes.
+ for dataframe in batch:
+ if dataframe.shape[0] != 1:
+ raise ValueError('Only dataframes with single rows are supported.')
+
+ # vectorize data for better performance
+ vectorized_batch = pandas.concat(batch, axis=0)
+ predictions = model.predict(vectorized_batch)
+ splits = [
+ vectorized_batch.iloc[[i]] for i in range(vectorized_batch.shape[0])
+ ]
+ return [
+ PredictionResult(example, inference) for example,
+ inference in zip(splits, predictions)
+ ]
+
+ def get_num_bytes(
+ self, batch: List[Union[numpy.ndarray, pandas.DataFrame]]) -> int:
"""Returns the number of bytes of data for a batch."""
- return sum(sys.getsizeof(element) for element in batch)
+ if isinstance(batch[0], numpy.ndarray):
+ return sum(sys.getsizeof(element) for element in batch)
+ elif isinstance(batch[0], pandas.DataFrame):
+ data_frames: List[pandas.DataFrame] = batch
+ return sum(df.memory_usage(deep=True).sum() for df in data_frames)
+ raise ValueError('Unsupported data type.')
class SklearnModelLoader(ModelLoader[numpy.ndarray,
Review Comment:
I think you missed this typehint
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] asf-ci commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
asf-ci commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1143888748
Can one of the admins verify this patch?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] asf-ci commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
asf-ci commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1143888746
Can one of the admins verify this patch?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] asf-ci commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
asf-ci commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1143888756
Can one of the admins verify this patch?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] yeandy commented on a diff in pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
yeandy commented on code in PR #17800:
URL: https://github.com/apache/beam/pull/17800#discussion_r891370951
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -71,7 +72,7 @@ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
# vectorize data for better performance
vectorized_batch = pandas.concat(batch, axis=0)
predictions = model.predict(vectorized_batch)
- splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
+ splits = [vectorized_batch.iloc[[i]] for i in range(len(batch))]
Review Comment:
What if the user accidentally passes in a DataFrame that has more than 1 row for an element in `batch`? i.e. [DF1(2 rows), DF2(2 rows), etc.]. Then indexing into `vectorized_batch.iloc[[i]]` will not yield what we assume is the first (and only) row from each DataFrame in `batch`. Instead, it will just take the first n rows of `vectorized_batch`, which is just the rows from DF1.
Should we have a check for this and throw an error for now?
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -56,6 +56,7 @@ def run_inference(
return SklearnInferenceRunner._predict_np_array(batch, model)
elif isinstance(batch[0], pandas.DataFrame):
return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
+ raise ValueError('Unsupported data type.')
Review Comment:
```suggestion
raise ValueError('Unsupported data type. Please use either numpy ndarray or pandas DataFrame')
```
And can we also add a test for this?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] asf-ci commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
asf-ci commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1143888745
Can one of the admins verify this patch?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1143940303
# [Codecov](https://codecov.io/gh/apache/beam/pull/17800?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#17800](https://codecov.io/gh/apache/beam/pull/17800?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (9fb4441) into [master](https://codecov.io/gh/apache/beam/commit/ca33943808c56ce634c92eb85f865285c71ee048?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ca33943) will **increase** coverage by `0.00%`.
> The diff coverage is `100.00%`.
```diff
@@ Coverage Diff @@
## master #17800 +/- ##
=======================================
Coverage 74.00% 74.01%
=======================================
Files 695 695
Lines 91898 91915 +17
=======================================
+ Hits 68013 68027 +14
- Misses 22639 22642 +3
Partials 1246 1246
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `83.74% <100.00%> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/17800?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...thon/apache\_beam/ml/inference/sklearn\_inference.py](https://codecov.io/gh/apache/beam/pull/17800/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL3NrbGVhcm5faW5mZXJlbmNlLnB5) | `94.82% <100.00%> (+2.14%)` | :arrow_up: |
| [sdks/python/apache\_beam/io/source\_test\_utils.py](https://codecov.io/gh/apache/beam/pull/17800/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vc291cmNlX3Rlc3RfdXRpbHMucHk=) | `88.01% <0.00%> (-1.39%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://codecov.io/gh/apache/beam/pull/17800/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `88.94% <0.00%> (-0.16%)` | :arrow_down: |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://codecov.io/gh/apache/beam/pull/17800/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `93.54% <0.00%> (-0.13%)` | :arrow_down: |
| [sdks/python/apache\_beam/ml/inference/api.py](https://codecov.io/gh/apache/beam/pull/17800/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL2FwaS5weQ==) | `100.00% <0.00%> (+10.00%)` | :arrow_up: |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/17800?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/17800?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [ca33943...9fb4441](https://codecov.io/gh/apache/beam/pull/17800?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit merged pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
TheNeuralBit merged PR #17800:
URL: https://github.com/apache/beam/pull/17800
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ryanthompson591 commented on a diff in pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
ryanthompson591 commented on code in PR #17800:
URL: https://github.com/apache/beam/pull/17800#discussion_r888238813
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
+
+ @staticmethod
+ def _predict_np_array(batch: List[numpy.ndarray],
+ model: Any) -> Iterable[PredictionResult]:
# vectorize data for better performance
vectorized_batch = numpy.stack(batch, axis=0)
predictions = model.predict(vectorized_batch)
return [PredictionResult(x, y) for x, y in zip(batch, predictions)]
- def get_num_bytes(self, batch: List[numpy.ndarray]) -> int:
+ @staticmethod
+ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
+ model: Any) -> Iterable[PredictionResult]:
+ # vectorize data for better performance
+ vectorized_batch = pandas.concat(batch, axis=0)
+ predictions = model.predict(vectorized_batch)
+ splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
+ return [
+ PredictionResult(example, inference) for example,
+ inference in zip(splits, predictions)
+ ]
+
+ def get_num_bytes(
+ self, batch: List[Union[numpy.ndarray, pandas.DataFrame]]) -> int:
"""Returns the number of bytes of data for a batch."""
- return sum(sys.getsizeof(element) for element in batch)
+ if isinstance(batch[0], numpy.ndarray):
+ return sum(sys.getsizeof(element) for element in batch)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return sum(df.memory_usage(deep=True).sum() for df in batch)
Review Comment:
Deep will introspect the data deeply by interrogating object dtypes for system-level memory consumption, and include it in the returned values.
##########
sdks/python/apache_beam/ml/inference/sklearn_inference_test.py:
##########
@@ -172,6 +231,53 @@ def test_bad_input_type_raises(self):
model_uri=file.name, model_file_type=None)
model_loader.load_model()
+ @unittest.skipIf(platform.system() == 'Windows', 'BEAM-14359')
+ def test_pipeline_pandas(self):
+ temp_file_name = self.tmpdir + os.sep + 'pickled_file'
+ with open(temp_file_name, 'wb') as file:
+ pickle.dump(build_pandas_pipeline(), file)
+ with TestPipeline() as pipeline:
+ data_frame = pandas_dataframe()
+
+ pcoll = pipeline | 'start' >> beam.Create([data_frame])
+ actual = pcoll | api.RunInference(
+ SklearnModelLoader(model_uri=temp_file_name))
+
+ splits = [data_frame.loc[[i]] for i in data_frame.index]
+ expected = [
+ api.PredictionResult(splits[0], 5),
+ api.PredictionResult(splits[1], 8),
+ api.PredictionResult(splits[2], 1),
+ api.PredictionResult(splits[3], 1),
+ api.PredictionResult(splits[4], 2),
+ ]
+ assert_that(
+ actual, equal_to(expected, equals_fn=_compare_dataframe_predictions))
+
+ @unittest.skipIf(platform.system() == 'Windows', 'BEAM-14359')
+ def test_pipeline_pandas_with_keys(self):
Review Comment:
I wanted it for this case, because I'm worried about a regression, because dataframes might be stacked or of different inputs.
##########
sdks/python/apache_beam/ml/inference/sklearn_inference_test.py:
##########
@@ -62,6 +83,44 @@ def build_model():
return model
+def pandas_dataframe():
+ csv_string = (
+ 'category_1,number_1,category_2,number_2,label,number_3\n'
Review Comment:
I've decided to not look closely at user defined indexes and instead just use the row index.
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
+
+ @staticmethod
+ def _predict_np_array(batch: List[numpy.ndarray],
+ model: Any) -> Iterable[PredictionResult]:
# vectorize data for better performance
vectorized_batch = numpy.stack(batch, axis=0)
predictions = model.predict(vectorized_batch)
return [PredictionResult(x, y) for x, y in zip(batch, predictions)]
- def get_num_bytes(self, batch: List[numpy.ndarray]) -> int:
+ @staticmethod
+ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
+ model: Any) -> Iterable[PredictionResult]:
+ # vectorize data for better performance
+ vectorized_batch = pandas.concat(batch, axis=0)
+ predictions = model.predict(vectorized_batch)
+ splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
Review Comment:
I think this is actually a bug that I have.
I made a change to account for this. Basically take your code and add this line:
vectorized_batch.reset_index(inplace=True, drop=True)
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
Review Comment:
yeah, I was thinking about your PR and think this sort of refactor would also look nice if it worked out.
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
Review Comment:
I don't know if one is used more than the other so I just went alphabetical.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ryanthompson591 commented on a diff in pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
ryanthompson591 commented on code in PR #17800:
URL: https://github.com/apache/beam/pull/17800#discussion_r893599417
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -41,20 +43,57 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
def run_inference(
- self, batch: List[numpy.ndarray], model: BaseEstimator,
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator,
**kwargs) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
Review Comment:
There was a conversation with heejong and the xlang folk as well as a separate one with the TFMA group going on about setting the input type correctly.
I think that that sort of a feature and a set_input_type would go hand in hand well.
Instead of implementing that in this PR, I think about making a more clean set_input_type PR that makes this more generic.
Let me leave an issue for it here:
https://github.com/apache/beam/issues/21769
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] yeandy commented on a diff in pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
yeandy commented on code in PR #17800:
URL: https://github.com/apache/beam/pull/17800#discussion_r887267918
##########
sdks/python/apache_beam/ml/inference/sklearn_inference_test.py:
##########
@@ -172,6 +231,53 @@ def test_bad_input_type_raises(self):
model_uri=file.name, model_file_type=None)
model_loader.load_model()
+ @unittest.skipIf(platform.system() == 'Windows', 'BEAM-14359')
+ def test_pipeline_pandas(self):
+ temp_file_name = self.tmpdir + os.sep + 'pickled_file'
+ with open(temp_file_name, 'wb') as file:
+ pickle.dump(build_pandas_pipeline(), file)
+ with TestPipeline() as pipeline:
+ data_frame = pandas_dataframe()
Review Comment:
nit
```suggestion
dataframe = pandas_dataframe()
```
##########
sdks/python/apache_beam/ml/inference/sklearn_inference_test.py:
##########
@@ -172,6 +231,53 @@ def test_bad_input_type_raises(self):
model_uri=file.name, model_file_type=None)
model_loader.load_model()
+ @unittest.skipIf(platform.system() == 'Windows', 'BEAM-14359')
+ def test_pipeline_pandas(self):
+ temp_file_name = self.tmpdir + os.sep + 'pickled_file'
+ with open(temp_file_name, 'wb') as file:
+ pickle.dump(build_pandas_pipeline(), file)
+ with TestPipeline() as pipeline:
+ data_frame = pandas_dataframe()
+
+ pcoll = pipeline | 'start' >> beam.Create([data_frame])
+ actual = pcoll | api.RunInference(
+ SklearnModelLoader(model_uri=temp_file_name))
+
+ splits = [data_frame.loc[[i]] for i in data_frame.index]
+ expected = [
+ api.PredictionResult(splits[0], 5),
+ api.PredictionResult(splits[1], 8),
+ api.PredictionResult(splits[2], 1),
+ api.PredictionResult(splits[3], 1),
+ api.PredictionResult(splits[4], 2),
+ ]
+ assert_that(
+ actual, equal_to(expected, equals_fn=_compare_dataframe_predictions))
+
+ @unittest.skipIf(platform.system() == 'Windows', 'BEAM-14359')
+ def test_pipeline_pandas_with_keys(self):
Review Comment:
We're already have a keyed test in `base_test.py`. Is it necessary to test a keyed example for each framework?
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
Review Comment:
I don't think there's a convention pertaining the order of elements in a Union? Since `pandas.DataFrame` is used more, would this be more useful?
```suggestion
batch: List[Union[pandas.DataFrame, numpy.ndarray]],
```
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
+
+ @staticmethod
+ def _predict_np_array(batch: List[numpy.ndarray],
+ model: Any) -> Iterable[PredictionResult]:
# vectorize data for better performance
vectorized_batch = numpy.stack(batch, axis=0)
predictions = model.predict(vectorized_batch)
return [PredictionResult(x, y) for x, y in zip(batch, predictions)]
- def get_num_bytes(self, batch: List[numpy.ndarray]) -> int:
+ @staticmethod
+ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
+ model: Any) -> Iterable[PredictionResult]:
+ # vectorize data for better performance
+ vectorized_batch = pandas.concat(batch, axis=0)
+ predictions = model.predict(vectorized_batch)
+ splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
Review Comment:
I did a toy example like so:
```
>>> df1 = pd.DataFrame([[1,2,3],[4,5,6]])
>>> df2 = pd.DataFrame([[11,12,13],[14,15,16]])
>>> vectorized_batch = pd.concat([df1, df2])
>>> vectorized_batch.index
Int64Index([0, 1, 0, 1], dtype='int64')
>>> vectorized_batch.loc[[0]]
0 1 2
0 1 2 3
0 11 12 13
>>> vectorized_batch.loc[[0]]
0 1 2
1 4 5 6
1 14 15 16
>>> splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
>>> splits
[ 0 1 2
0 1 2 3
0 11 12 13, 0 1 2
1 4 5 6
1 14 15 16, 0 1 2
0 1 2 3
0 11 12 13, 0 1 2
1 4 5 6
1 14 15 16]
```
I see that index 0 is found in both the first and second batch. Same for index 1. So indexing by `vectorized_batch.index` appears to up data from the different DataFrames.
Can you clarify how `vectorized_batch.index` is supposed to differentiate between the batches?
##########
sdks/python/apache_beam/ml/inference/sklearn_inference_test.py:
##########
@@ -62,6 +83,44 @@ def build_model():
return model
+def pandas_dataframe():
+ csv_string = (
+ 'category_1,number_1,category_2,number_2,label,number_3\n'
Review Comment:
Does this implicitly add indexes starting from 0? What if the user has custom indexes?
##########
sdks/python/apache_beam/ml/inference/sklearn_inference_test.py:
##########
@@ -172,6 +231,53 @@ def test_bad_input_type_raises(self):
model_uri=file.name, model_file_type=None)
model_loader.load_model()
+ @unittest.skipIf(platform.system() == 'Windows', 'BEAM-14359')
+ def test_pipeline_pandas(self):
+ temp_file_name = self.tmpdir + os.sep + 'pickled_file'
+ with open(temp_file_name, 'wb') as file:
+ pickle.dump(build_pandas_pipeline(), file)
+ with TestPipeline() as pipeline:
+ data_frame = pandas_dataframe()
+
+ pcoll = pipeline | 'start' >> beam.Create([data_frame])
Review Comment:
So here, we have a list of 1 DataFrame. Can we test a list of 2 or more DataFrames which would simulate reading in separate text files into their own DFs?
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
Review Comment:
Nice refactor
##########
sdks/python/apache_beam/ml/inference/sklearn_inference.py:
##########
@@ -42,19 +44,46 @@ class ModelFileType(enum.Enum):
JOBLIB = 2
-class SklearnInferenceRunner(InferenceRunner[numpy.ndarray,
+class SklearnInferenceRunner(InferenceRunner[Union[numpy.ndarray,
+ pandas.DataFrame],
PredictionResult,
BaseEstimator]):
- def run_inference(self, batch: List[numpy.ndarray],
- model: BaseEstimator) -> Iterable[PredictionResult]:
+ def run_inference(
+ self,
+ batch: List[Union[numpy.ndarray, pandas.DataFrame]],
+ model: BaseEstimator) -> Iterable[PredictionResult]:
+ if isinstance(batch[0], numpy.ndarray):
+ return SklearnInferenceRunner._predict_np_array(batch, model)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return SklearnInferenceRunner._predict_pandas_dataframe(batch, model)
+
+ @staticmethod
+ def _predict_np_array(batch: List[numpy.ndarray],
+ model: Any) -> Iterable[PredictionResult]:
# vectorize data for better performance
vectorized_batch = numpy.stack(batch, axis=0)
predictions = model.predict(vectorized_batch)
return [PredictionResult(x, y) for x, y in zip(batch, predictions)]
- def get_num_bytes(self, batch: List[numpy.ndarray]) -> int:
+ @staticmethod
+ def _predict_pandas_dataframe(batch: List[pandas.DataFrame],
+ model: Any) -> Iterable[PredictionResult]:
+ # vectorize data for better performance
+ vectorized_batch = pandas.concat(batch, axis=0)
+ predictions = model.predict(vectorized_batch)
+ splits = [vectorized_batch.loc[[i]] for i in vectorized_batch.index]
+ return [
+ PredictionResult(example, inference) for example,
+ inference in zip(splits, predictions)
+ ]
+
+ def get_num_bytes(
+ self, batch: List[Union[numpy.ndarray, pandas.DataFrame]]) -> int:
"""Returns the number of bytes of data for a batch."""
- return sum(sys.getsizeof(element) for element in batch)
+ if isinstance(batch[0], numpy.ndarray):
+ return sum(sys.getsizeof(element) for element in batch)
+ elif isinstance(batch[0], pandas.DataFrame):
+ return sum(df.memory_usage(deep=True).sum() for df in batch)
Review Comment:
What does `deep=True` mean?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ryanthompson591 commented on pull request #17800: [BEAM-14535] Added support for pandas in sklearn inference runner
Posted by GitBox <gi...@apache.org>.
ryanthompson591 commented on PR #17800:
URL: https://github.com/apache/beam/pull/17800#issuecomment-1143903311
R: @yeandy
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org