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.

Sign up for GitHub

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

Jump to bottom

[AIR] Fix examples and tests to specify batch_format in BatchMapper #30438

Merged
merged 2 commits into from
Nov 18, 2022
Merged

[AIR] Fix examples and tests to specify batch_format in BatchMapper #30438

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c889276
Add batch_format to noop BatchMapper in batch predictor logic
justinvyu Nov 17, 2022
eec712e
Update all examples/tests that use BatchMapper to specify batch_format
justinvyu Nov 17, 2022
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
12 changes: 6 additions & 6 deletions doc/source/ray-air/doc_code/air_ingest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@

# An example preprocessor chain that just scales all values by 4.0 in two stages.
preprocessor = Chain(
BatchMapper(lambda df: df * 2),
BatchMapper(lambda df: df * 2),
BatchMapper(lambda df: df * 2, batch_format="pandas"),
BatchMapper(lambda df: df * 2, batch_format="pandas"),
)
# __check_ingest_1_end__

Expand Down Expand Up @@ -93,7 +93,7 @@
from ray.air.config import ScalingConfig

# A simple preprocessor that just scales all values by 2.0.
preprocessor = BatchMapper(lambda df: df * 2)
preprocessor = BatchMapper(lambda df: df * 2, batch_format="pandas")


def train_loop_per_worker():
Expand Down Expand Up @@ -128,7 +128,7 @@ def train_loop_per_worker():
from ray.air.config import ScalingConfig, DatasetConfig

# A simple preprocessor that just scales all values by 2.0.
preprocessor = BatchMapper(lambda df: df * 2)
preprocessor = BatchMapper(lambda df: df * 2, batch_format="pandas")


def train_loop_per_worker():
Expand Down Expand Up @@ -253,7 +253,7 @@ def train_loop_per_worker():


# A simple preprocessor that just scales all values by 2.0.
preprocessor = BatchMapper(lambda df: df * 2)
preprocessor = BatchMapper(lambda df: df * 2, batch_format="pandas")

my_trainer = TorchTrainer(
train_loop_per_worker,
Expand Down Expand Up @@ -290,7 +290,7 @@ def train_loop_per_worker():


# A simple preprocessor that just scales all values by 2.0.
preprocessor = BatchMapper(lambda df: df * 2)
preprocessor = BatchMapper(lambda df: df * 2, batch_format="pandas")

my_trainer = TorchTrainer(
train_loop_per_worker,
Expand Down
2 changes: 1 addition & 1 deletion doc/source/ray-air/doc_code/preprocessors.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@
# [{'value': 0}, {'value': 1}, {'value': 2}, {'value': 3}]

# Create a stateless preprocess that multiplies values by 2.
preprocessor = BatchMapper(lambda df: df * 2, batch_size=2)
preprocessor = BatchMapper(lambda df: df * 2, batch_size=2, batch_format="pandas")
dataset_transformed = preprocessor.transform(dataset)
print(dataset_transformed.take())
# [{'value': 0}, {'value': 2}, {'value': 4}, {'value': 6}]
Expand Down
8 changes: 4 additions & 4 deletions doc/source/ray-air/examples/huggingface_text_classification.ipynb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,7 +500,7 @@
" ret = {**examples, **ret}\n",
" return pd.DataFrame.from_dict(ret)\n",
"\n",
"batch_encoder = BatchMapper(preprocess_function)"
"batch_encoder = BatchMapper(preprocess_function, batch_format=\"pandas\")"
]
},
{
Expand Down Expand Up @@ -2255,7 +2255,7 @@
"provenance": []
},
"kernelspec": {
"display_name": "Python 3.8.10 ('venv': venv)",
"display_name": "Python 3.8.9 64-bit",
"language": "python",
"name": "python3"
},
Expand All @@ -2269,11 +2269,11 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.8.10"
"version": "3.8.9"
},
"vscode": {
"interpreter": {
"hash": "3c0d54d489a08ae47a06eae2fd00ff032d6cddb527c382959b7b2575f6a8167f"
"hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
}
}
},
Expand Down
10 changes: 5 additions & 5 deletions doc/source/ray-air/examples/tfx_tabular_train_to_serve.ipynb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -587,8 +587,8 @@
" imputer4,\n",
" imputer5,\n",
" ohe,\n",
" BatchMapper(batch_mapper_fn),\n",
" BatchMapper(concat_for_tensor)\n",
" BatchMapper(batch_mapper_fn, batch_format=\"pandas\"),\n",
" BatchMapper(concat_for_tensor, batch_format=\"pandas\")\n",
" )\n",
" return chained_pp\n"
]
Expand Down Expand Up @@ -1888,7 +1888,7 @@
"provenance": []
},
"kernelspec": {
"display_name": "Python 3.10.8 ('.venv': venv)",
"display_name": "Python 3.8.9 64-bit",
"language": "python",
"name": "python3"
},
Expand All @@ -1902,11 +1902,11 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.8"
"version": "3.8.9"
},
"vscode": {
"interpreter": {
"hash": "c704e19737f24b51bc631dadcac7a7e356bb35d1c5cd7766248d8a6946059909"
"hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
}
}
},
Expand Down
6 changes: 3 additions & 3 deletions python/ray/air/tests/test_dataset_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ def test_error(ray_start_4_cpus):
{"train": 10},
dataset_config={},
datasets={"train": ds},
preprocessor=BatchMapper(lambda x: x),
preprocessor=BatchMapper(lambda x: x, batch_format="pandas"),
)
test.fit()

Expand Down Expand Up @@ -260,7 +260,7 @@ def checker(shard, results):
def rand(x):
return [random.random() for _ in range(len(x))]

prep = BatchMapper(rand)
prep = BatchMapper(rand, batch_format="pandas")
ds = ray.data.range_table(5, parallelism=1)
test = TestStream(
checker,
Expand All @@ -275,7 +275,7 @@ def test_stream_finite_window_nocache_prep(ray_start_4_cpus):
def rand(x):
return [random.random() for _ in range(len(x))]

prep = BatchMapper(rand)
prep = BatchMapper(rand, batch_format="pandas")
ds = ray.data.range_table(5)

# Test the default 1GiB window size.
Expand Down
4 changes: 2 additions & 2 deletions python/ray/air/util/check_ingest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,8 +163,8 @@ def generate_epochs(data: Union[Dataset, DatasetPipeline], epochs: int):

# An example preprocessor chain that just scales all values by 4.0 in two stages.
preprocessor = Chain(
BatchMapper(lambda df: df * 2),
BatchMapper(lambda df: df * 2),
BatchMapper(lambda df: df * 2, batch_format="pandas"),
BatchMapper(lambda df: df * 2, batch_format="pandas"),
)

# Setup the dummy trainer that prints ingest stats.
Expand Down
4 changes: 3 additions & 1 deletion python/ray/train/batch_predictor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,7 +287,9 @@ def __call__(self, input_batch: DataBatchType) -> DataBatchType:
if preprocessor:
# Set the in-predictor preprocessing to a no-op when using a separate
# GPU stage. Otherwise, the preprocessing will be applied twice.
override_prep = BatchMapper(lambda x: x)
override_prep = BatchMapper(
lambda x: x, batch_format=predict_stage_batch_format
)
Copy link
Contributor Author

Choose a reason for hiding this comment

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

Is it possible to add a unit test for this codepath? Test preprocessing before sending to a separate GPU stage to do prediction.

Copy link
Contributor

Choose a reason for hiding this comment

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

this is being tested here https://sourcegraph.com/github.com/ray-project/ray/-/blob/python/ray/train/tests/test_batch_predictor.py?L104:5

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I see, I think the test was still pending and not showing up on the flaky tests earlier.

# preprocessor.transform will break for DatasetPipeline due to
# missing _dataset_format()
batch_fn = preprocessor._transform_batch
Expand Down
2 changes: 1 addition & 1 deletion python/ray/train/tests/test_base_trainer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,7 +407,7 @@ def map_fn(batch):
assert ctx.target_max_block_size == target_max_block_size
return batch + 1

preprocessor = BatchMapper(map_fn)
preprocessor = BatchMapper(map_fn, batch_format="pandas")

ctx = ray.data.context.DatasetContext.get_current()
ctx.target_max_block_size = target_max_block_size
Expand Down
10 changes: 6 additions & 4 deletions release/air_tests/air_benchmarks/mlperf-train/resnet50_ray_air.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,7 +221,7 @@ def crop_and_flip_image_batch(image_batch):
return image_batch


def decode_tf_record_batch(tf_record_batch):
def decode_tf_record_batch(tf_record_batch: pd.DataFrame) -> pd.DataFrame:
def process_images():
for image_buffer in tf_record_batch["image/encoded"]:
image_buffer = tf.reshape(image_buffer, shape=[])
Expand All @@ -237,7 +237,7 @@ def process_images():
return df


def decode_crop_and_flip_tf_record_batch(tf_record_batch):
def decode_crop_and_flip_tf_record_batch(tf_record_batch: pd.DataFrame) -> pd.DataFrame:
"""
This version of the preprocessor fuses the load step with the crop and flip
step, which should have better performance (at the cost of re-executing the
Expand Down Expand Up @@ -518,11 +518,13 @@ def append_to_test_output_json(path, metrics):
batch_size = 32
if args.online_processing:
preprocessor = BatchMapper(
decode_tf_record_batch, batch_size=batch_size
decode_tf_record_batch, batch_size=batch_size, batch_format="pandas"
)
else:
preprocessor = BatchMapper(
decode_crop_and_flip_tf_record_batch, batch_size=batch_size
decode_crop_and_flip_tf_record_batch,
batch_size=batch_size,
batch_format="pandas",
)
train_loop_config["data_loader"] = RAY_DATA

Expand Down
2 changes: 1 addition & 1 deletion release/air_tests/air_benchmarks/workloads/data_benchmark.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ def make_ds(size_gb: int):


def run_ingest_bulk(dataset, num_workers, num_cpus_per_worker):
dummy_prep = BatchMapper(lambda df: df * 2)
dummy_prep = BatchMapper(lambda df: df * 2, batch_format="pandas")
trainer = DummyTrainer(
scaling_config=ScalingConfig(
num_workers=num_workers,
Expand Down