damccorm commented on code in PR #41200:
URL: https://github.com/apache/beam/pull/41200#discussion_r4540675207
##########
sdks/python/apache_beam/examples/inference/table_row_inference.py:
##########
@@ -282,6 +283,21 @@ def run(
'In batch mode, specify at least one of --output_table or
--output_file'
)
+ # Create the Pub/Sub topic and subscription before submit so Dataflow
+ # preflight validation finds them.
+ if known_args.mode == 'streaming':
+ parts = known_args.input_subscription.strip('/').split('/')
+ if (len(parts) != 4 or parts[0] != 'projects' or
+ parts[2] != 'subscriptions'):
+ raise ValueError(
+ 'input_subscription must be projects/PROJECT/subscriptions/NAME')
+ project = parts[1]
+ subscription_name = parts[3]
+ topic_name = subscription_name
+ table_row_inference_utils.ensure_pubsub_topic(project, topic_name)
+ table_row_inference_utils.ensure_pubsub_subscription(
+ project, topic_name, subscription_name)
Review Comment:
In streaming mode (`known_args.mode == "streaming"`),
`table_row_inference.run()` returns `result = pipeline.run()` immediately
without waiting/cancelling, and `DataflowCostBenchmark.run()` only calls
`self.result.wait_until_finish(duration=self.timeout_ms)` (which returns
`PipelineState.RUNNING` after 15 minutes without cancelling the job). Unlike
the PyTorch streaming benchmarks, nothing cancels the streaming Dataflow job or
cleans up the temporary topic/subscription, leaving
`benchmark-tests-table-row-inference-stream-*` running indefinitely on Dataflow
after the GitHub Action completes. Also, nothing publishes data to the newly
created `table_row_inference_benchmark` topic during the run.
Could we follow a similar pattern to `pytorch_image_captioning.py` /
`pytorch_imagenet_rightfit.py` (publish test data to the topic, wait for the
benchmark duration, and `result.cancel()` + clean up Pub/Sub resources in a
`finally:` block) so the streaming job does not leak?
##########
.github/workflows/beam_Inference_Python_Benchmarks_Dataflow.yml:
##########
@@ -348,9 +348,9 @@ jobs:
arguments: |
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_imagenet_rightfit_benchmarks
\
-Prunner=DataflowRunner \
- -PpythonVersion=3.11 \
+ -PpythonVersion=3.12 \
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_rightfit_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_11 }} --device=GPU
--experiments=enable_streaming_rightfitting,worker_accelerator=type:nvidia-tesla-t4;count:1;install-nvidia-driver:5xx
--mode=streaming
--job_name=benchmark-tests-pytorch-imagenet-rightfit-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_imagenet_stream_rightfit_once_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_imagenet_stream_rightfit_once_gpu'
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_11 }} --device=GPU
--experiments=enable_streaming_rightfitting,worker_accelerator=type:nvidia-tesla-t4;count:1;install-nvidia-driver:5xx
--sdk_container_image=us.gcr.io/apache-beam-testing/python-postcommit-it/tensor_rt:latest
--num_workers=30 --disk_size_gb=100 --mode=streaming
--job_name=benchmark-tests-pytorch-imagenet-rightfit-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_imagenet_stream_rightfit_once_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_imagenet_stream_rightfit_once_gpu'
\
Review Comment:
Nit: Both this step (`Right-fitting Exactly-once`) and the preceding step
(`Right-fitting` at line 342) use the same
`--job_name=benchmark-tests-pytorch-imagenet-rightfit-streaming-${{env.NOW_UTC}}`
(where `NOW_UTC` is evaluated once at workflow startup). Giving this step a
distinct job name (e.g.
`benchmark-tests-pytorch-imagenet-rightfit-eo-streaming-${{env.NOW_UTC}}`)
would make the two jobs easier to distinguish in the Dataflow console.
##########
.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Captioning.txt:
##########
@@ -15,6 +15,7 @@
--region=us-central1
--worker_machine_type=n1-standard-4
--num_workers=50
+--timeout_ms=600000
Review Comment:
It looks like `--timeout_ms=600000` (10 min) was added here (and in
`Pytorch_Image_Classification_Rightfit.txt` and
`Pytorch_Image_Object_Detection.txt`) to bound the streaming benchmark runtime,
but `pytorch_image_captioning.py`, `pytorch_image_object_detection.py`, and
`pytorch_imagenet_rightfit.py` hardcode
`result.wait_until_finish(duration=1800000)` (30 min) inside `run()` before
returning `result` to `DataflowCostBenchmark`. By the time
`DataflowCostBenchmark` checks `self.timeout_ms`, the job has already run for
30 minutes and been cancelled.
If we want `--timeout_ms` to control the streaming duration, we should wire
it through to `run()`; otherwise, if 30 minutes is intentional, we may want to
remove the no-op `--timeout_ms=600000` from these 3 option files to avoid
confusion.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
