This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new c102eb4c321 ML pipelines: RunInference - OSS Image Object detection,
OSS Image Captioning, OSS Image Classification (#37186)
c102eb4c321 is described below
commit c102eb4c32169d724c1d51a4f36e2925a0714203
Author: Vitaly Terentyev <[email protected]>
AuthorDate: Fri Jul 17 15:49:23 2026 +0400
ML pipelines: RunInference - OSS Image Object detection, OSS Image
Captioning, OSS Image Classification (#37186)
* ML pipelines: RunInference - OSS Image Object detection
* Refactoring
* Add PyTorch Image Captioning BLIP + CLIP Batch
* Refactoring
* Add PyTorch Image Classification EfficientNet-B0 Streaming
(Right-fitting) ML pipeline
* Fix lint
* Rename files
* Add project argument
* Fix inference_fn
* Fix inference result
* Update torch for image captioning
* Fix inference
* Add Looker ids
* Fix float for imagenet rightfit
* Add all pipelines to separate workflow
* Fix streaming mode
* Fix clip blip
* Fix quotes
* Fix formatting
* Fix lint
* Add Looker IDs
* Resolve comments [1]
* Resolve comments [2]
* Resolve comments [3]
* Resolve comments [4]
* Fix conflicts
* Fix formatting
* Remove rate limit
* Refactor model and process init
* Add beam_Inference_Python_Benchmarks_Dataflow_2 to README.md
* Refactoring
* Move to one yml file
* Resolve comments
* Resolve conflicts
* Resolve comments
* Fix formatter
* Resolve comments
* Change default batch size
* Fix formatting
---
.../beam_Inference_Python_Benchmarks_Dataflow.yml | 153 ++++-
...enchmarks_Dataflow_Pytorch_Image_Captioning.txt | 37 ++
...aflow_Pytorch_Image_Classification_Rightfit.txt | 41 ++
...rks_Dataflow_Pytorch_Image_Object_Detection.txt | 37 ++
.test-infra/tools/refresh_looker_metrics.py | 19 +
.../examples/inference/pytorch_image_captioning.py | 641 +++++++++++++++++++++
.../inference/pytorch_image_object_detection.py | 517 +++++++++++++++++
.../inference/pytorch_imagenet_rightfit.py | 537 +++++++++++++++++
.../pytorch_image_captioning_requirements.txt | 27 +
...pytorch_image_object_detection_requirements.txt | 24 +
.../ml/inference/pytorch_rightfit_requirements.txt | 26 +
.../pytorch_image_captioning_benchmarks.py | 42 ++
.../pytorch_image_object_detection_benchmarks.py | 42 ++
.../pytorch_imagenet_rightfit_benchmarks.py | 42 ++
website/www/site/content/en/performance/_index.md | 12 +
.../pytorchimagecaptioningbatchcpu/_index.md | 41 ++
.../pytorchimagecaptioningbatchgpu/_index.md | 41 ++
.../pytorchimagecaptioningstreamingcpu/_index.md | 41 ++
.../pytorchimagecaptioningstreaminggpu/_index.md | 41 ++
.../pytorchimagenetrightfitcpu/_index.md | 43 ++
.../pytorchimagenetrightfitgpu/_index.md | 43 ++
.../pytorchimagenetrightfitoncecpu/_index.md | 44 ++
.../pytorchimagenetrightfitoncegpu/_index.md | 44 ++
.../pytorchimageobjectdetectionbatchcpu/_index.md | 41 ++
.../pytorchimageobjectdetectionbatchgpu/_index.md | 41 ++
.../_index.md | 41 ++
.../_index.md | 41 ++
website/www/site/data/performance.yaml | 206 ++++++-
28 files changed, 2889 insertions(+), 16 deletions(-)
diff --git a/.github/workflows/beam_Inference_Python_Benchmarks_Dataflow.yml
b/.github/workflows/beam_Inference_Python_Benchmarks_Dataflow.yml
index ce3017f40b0..fb7fecd2c01 100644
--- a/.github/workflows/beam_Inference_Python_Benchmarks_Dataflow.yml
+++ b/.github/workflows/beam_Inference_Python_Benchmarks_Dataflow.yml
@@ -55,7 +55,7 @@ jobs:
(github.event_name == 'schedule' && github.repository == 'apache/beam')
||
github.event.comment.body == 'Run Inference Benchmarks'
runs-on: [self-hosted, ubuntu-24.04, main]
- timeout-minutes: 1000
+ timeout-minutes: 3000
name: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
strategy:
matrix:
@@ -96,6 +96,9 @@ jobs:
${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_VLLM_Gemma_Batch.txt
${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Table_Row_Inference_Batch.txt
${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Table_Row_Inference_Stream.txt
+ ${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Classification_Rightfit.txt
+ ${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Object_Detection.txt
+ ${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Captioning.txt
${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_MLTransform_Generate_Vocab_Batch.txt
${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_MLTransform_One_Hot_Encoding_Batch.txt
${{ github.workspace
}}/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_MLTransform_Image_Embedding_GPU_Batch.txt
@@ -198,7 +201,7 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/torch_tests_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_5 }}
--job_name=benchmark-tests-pytorch-imagenet-python-gpu-${{env.NOW_UTC}}
--output=gs://temp-storage-for-end-to-end-tests/torch/result_resnet152_gpu-${{env.NOW_UTC}}.txt'
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_5 }}
--job_name=benchmark-tests-pytorch-imagenet-python-gpu-${{env.NOW_UTC}}
--output=gs://temp-storage-for-end-to-end-tests/torch/result_resnet152_gpu-${{env.NOW_UTC}}.txt'
\
- name: run Table Row Inference Sklearn Batch
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -209,7 +212,7 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/table_row_inference_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_9 }}
--autoscaling_algorithm=NONE --metrics_table=result_table_row_inference_batch
--influx_measurement=result_table_row_inference_batch --mode=batch
--input_file=gs://apache-beam-ml/testing/inputs/table_rows_100k_benchmark.jsonl
--input_expand_factor=100
--output_table=apache-beam-testing:beam_run_inference.result_table_row_inference_batch_outputs
--job_name=benchmark-tests-table-row-inference- [...]
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_9 }}
--autoscaling_algorithm=NONE --metrics_table=result_table_row_inference_batch
--influx_measurement=result_table_row_inference_batch --mode=batch
--input_file=gs://apache-beam-ml/testing/inputs/table_rows_100k_benchmark.jsonl
--input_expand_factor=100
--output_table=apache-beam-testing:beam_run_inference.result_table_row_inference_batch_outputs
--job_name=benchmark-tests-table-row-inference- [...]
- name: run Table Row Inference Sklearn Stream
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -220,7 +223,139 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/table_row_inference_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_10 }}
--autoscaling_algorithm=THROUGHPUT_BASED --max_num_workers=20
--metrics_table=result_table_row_inference_stream
--influx_measurement=result_table_row_inference_stream --mode=streaming
--input_subscription=projects/apache-beam-testing/subscriptions/table_row_inference_benchmark
--window_size_sec=60 --trigger_interval_sec=30 --timeout_ms=900000
--output_table=apache-beam-testing:beam_run_inf [...]
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_10 }}
--autoscaling_algorithm=THROUGHPUT_BASED --max_num_workers=20
--metrics_table=result_table_row_inference_stream
--influx_measurement=result_table_row_inference_stream --mode=streaming
--input_subscription=projects/apache-beam-testing/subscriptions/table_row_inference_benchmark
--window_size_sec=60 --trigger_interval_sec=30 --timeout_ms=900000
--output_table=apache-beam-testing:beam_run_inf [...]
+ - name: run PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch
CPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_object_detection_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_12 }} --device=CPU
--mode=batch
--job_name=benchmark-tests-pytorch-image-object-detection-batch-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_object_detection_batch_cpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_object_detection_batch_cpu'
\
+ - name: run PyTorch Image Captioning BLIP + CLIP Batch CPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_captioning_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_13 }} --device=CPU
--mode=batch
--job_name=benchmark-tests-pytorch-image-captioning-batch-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_captioning_batch_cpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_captioning_batch_cpu'
+ - name: run PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch
GPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_object_detection_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_12 }} --device=GPU
--worker_accelerator=type=nvidia-tesla-t4,count=1,install-nvidia-driver=true
--mode=batch
--job_name=benchmark-tests-pytorch-image-object-detection-batch-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_object_detection_batch_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_object_detection_batch_gpu'
\
+ - name: run PyTorch Image Captioning BLIP + CLIP Batch GPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_captioning_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_13 }} --device=GPU
--worker_accelerator=type=nvidia-tesla-t4,count=1,install-nvidia-driver=true
--mode=batch
--job_name=benchmark-tests-pytorch-image-captioning-batch-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_captioning_batch_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_captioning_batch_gpu'
\
+ - name: run PyTorch Image Object Detection Faster R-CNN ResNet-50
Streaming CPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_object_detection_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_12 }} --device=CPU
--mode=streaming
--job_name=benchmark-tests-pytorch-image-object-detection-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_object_detection_streaming_cpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_object_detection_streaming_cpu'
\
+ - name: run PyTorch Image Captioning BLIP + CLIP Streaming CPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_captioning_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_13 }} --device=CPU
--mode=streaming
--job_name=benchmark-tests-pytorch-image-captioning-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_captioning_streaming_cpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_captioning_streaming_cpu'
+ - name: run PyTorch Image Object Detection Faster R-CNN ResNet-50
Streaming GPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_object_detection_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_12 }} --device=GPU
--worker_accelerator=type=nvidia-tesla-t4,count=1,install-nvidia-driver=true
--mode=streaming
--job_name=benchmark-tests-pytorch-image-object-detection-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_object_detection_streaming_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_object_detection_streaming_gpu'
\
+ - name: run PyTorch Image Captioning BLIP + CLIP Streaming GPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_image_captioning_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_13 }} --device=GPU
--worker_accelerator=type=nvidia-tesla-t4,count=1,install-nvidia-driver=true
--mode=streaming
--job_name=benchmark-tests-pytorch-image-captioning-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_image_captioning_streaming_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_image_captioning_streaming_gpu'
\
+ - name: run PyTorch Image Classification EfficientNet-B0 Streaming
(Right-fitting) CPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_imagenet_rightfit_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_rightfit_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_11 }} --device=CPU
--experiments=enable_streaming_rightfitting --mode=streaming
--job_name=benchmark-tests-pytorch-imagenet-rightfit-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_imagenet_stream_rightfit_cpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_imagenet_stream_rightfit_cpu'
\
+ - name: run PyTorch Image Classification EfficientNet-B0 Streaming
(Right-fitting Exactly-once) CPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_imagenet_rightfit_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-PloadTest.requirementsTxtFile=apache_beam/ml/inference/pytorch_rightfit_requirements.txt
\
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_11 }} --device=CPU
--experiments=enable_streaming_rightfitting --mode=streaming
--job_name=benchmark-tests-pytorch-imagenet-rightfit-streaming-${{env.NOW_UTC}}
--metrics_table=torch_inference_imagenet_stream_rightfit_once_cpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_imagenet_stream_rightfit_once_cpu'
\
+ - name: run PyTorch Image Classification EfficientNet-B0 Streaming
(Right-fitting) GPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_imagenet_rightfit_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-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_gpu
--output_table=apache-beam-testing.beam_run_inference.result_torch_inference_imagenet_stream_rightfit_gpu'
\
+ - name: run PyTorch Image Classification EfficientNet-B0 Streaming
(Right-fitting Exactly-once) GPU
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ timeout-minutes: 180
+ with:
+ gradle-command: :sdks:python:apache_beam:testing:load_tests:run
+ arguments: |
+
-PloadTest.mainClass=apache_beam.testing.benchmarks.inference.pytorch_imagenet_rightfit_benchmarks
\
+ -Prunner=DataflowRunner \
+ -PpythonVersion=3.10 \
+
-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_right
[...]
- name: run MLTransform Generate Vocab Batch
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -231,7 +366,7 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/examples/ml_transform/mltransform_generate_vocab_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_11 }}
--job_name=benchmark-tests-mltransform-generate-vocab-batch-${{env.NOW_UTC}}
--artifact_location=gs://temp-storage-for-perf-tests/mltransform/vocab_artifacts_${{env.NOW_UTC}}
--output_vocab=gs://temp-storage-for-perf-tests/mltransform/vocab_outputs/mltransform_generate_vocab_batch_${{env.NOW_UTC}}'
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_14 }}
--job_name=benchmark-tests-mltransform-generate-vocab-batch-${{env.NOW_UTC}}
--artifact_location=gs://temp-storage-for-perf-tests/mltransform/vocab_artifacts_${{env.NOW_UTC}}
--output_vocab=gs://temp-storage-for-perf-tests/mltransform/vocab_outputs/mltransform_generate_vocab_batch_${{env.NOW_UTC}}'
- name: run MLTransform One-Hot Encoding Batch
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -242,7 +377,7 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/examples/ml_transform/mltransform_one_hot_encoding_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_12 }}
--autoscaling_algorithm=NONE --metrics_table=mltransform_one_hot_encoding_batch
--influx_measurement=mltransform_one_hot_encoding_batch
--job_name=benchmark-tests-mltransform-one-hot-encoding-batch-${{env.NOW_UTC}}
--output_file=gs://temp-storage-for-end-to-end-tests/mltransform/one_hot_output_${{env.NOW_UTC}}
--artifact_location=gs://temp-storage-for-end-to-end-tests/mltransform/artifacts
[...]
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_15 }}
--autoscaling_algorithm=NONE --metrics_table=mltransform_one_hot_encoding_batch
--influx_measurement=mltransform_one_hot_encoding_batch
--job_name=benchmark-tests-mltransform-one-hot-encoding-batch-${{env.NOW_UTC}}
--output_file=gs://temp-storage-for-end-to-end-tests/mltransform/one_hot_output_${{env.NOW_UTC}}
--artifact_location=gs://temp-storage-for-end-to-end-tests/mltransform/artifacts
[...]
- name: run MLTransform Image Embedding GPU Batch
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -253,7 +388,7 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/ml/transforms/mltransform_embedding_tests_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_13 }}
--artifact_location=gs://temp-storage-for-perf-tests/loadtests/mltransform_image_embedding_gpu/artifacts-${{env.NOW_UTC}}
--job_name=benchmark-tests-mltransform-image-embedding-gpu-batch-${{env.NOW_UTC}}'
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_16 }}
--artifact_location=gs://temp-storage-for-perf-tests/loadtests/mltransform_image_embedding_gpu/artifacts-${{env.NOW_UTC}}
--job_name=benchmark-tests-mltransform-image-embedding-gpu-batch-${{env.NOW_UTC}}'
- name: run MLTransform Text Embedding Batch
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -264,7 +399,7 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/ml/transforms/mltransform_embedding_tests_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_14 }}
--output=gs://temp-storage-for-perf-tests/loadtests/mltransform_text_embedding/results-${{env.NOW_UTC}}
--artifact_location=gs://temp-storage-for-perf-tests/loadtests/mltransform_text_embedding/artifacts-${{env.NOW_UTC}}
--job_name=benchmark-tests-mltransform-text-embedding-batch-${{env.NOW_UTC}}'
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_17 }}
--output=gs://temp-storage-for-perf-tests/loadtests/mltransform_text_embedding/results-${{env.NOW_UTC}}
--artifact_location=gs://temp-storage-for-perf-tests/loadtests/mltransform_text_embedding/artifacts-${{env.NOW_UTC}}
--job_name=benchmark-tests-mltransform-text-embedding-batch-${{env.NOW_UTC}}'
- name: run MLTransform Image Embedding CPU Batch
uses: ./.github/actions/gradle-command-self-hosted-action
timeout-minutes: 180
@@ -275,4 +410,4 @@ jobs:
-Prunner=DataflowRunner \
-PpythonVersion=3.10 \
-PloadTest.requirementsTxtFile=apache_beam/ml/transforms/mltransform_embedding_tests_requirements.txt
\
- '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_15 }}
--artifact_location=gs://temp-storage-for-perf-tests/loadtests/mltransform_image_embedding_cpu/artifacts-${{env.NOW_UTC}}
--job_name=benchmark-tests-mltransform-image-embedding-cpu-batch-${{env.NOW_UTC}}'
+ '-PloadTest.args=${{
env.beam_Inference_Python_Benchmarks_Dataflow_test_arguments_18 }}
--artifact_location=gs://temp-storage-for-perf-tests/loadtests/mltransform_image_embedding_cpu/artifacts-${{env.NOW_UTC}}
--job_name=benchmark-tests-mltransform-image-embedding-cpu-batch-${{env.NOW_UTC}}'
diff --git
a/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Captioning.txt
b/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Captioning.txt
new file mode 100644
index 00000000000..c58707383ef
--- /dev/null
+++
b/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Captioning.txt
@@ -0,0 +1,37 @@
+# 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.
+
+--region=us-central1
+--worker_machine_type=n1-standard-4
+--num_workers=50
+--disk_size_gb=50
+--autoscaling_algorithm=NONE
+--staging_location=gs://temp-storage-for-perf-tests/loadtests
+--temp_location=gs://temp-storage-for-perf-tests/loadtests
+--requirements_file=apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
+--publish_to_big_query=true
+--metrics_dataset=beam_run_inference
+--input_options={}
+--influx_measurement=result_torch_inference_image_captioning
+--input=gs://apache-beam-ml/testing/inputs/openimage_50k_benchmark.txt
+--blip_model_name=Salesforce/blip-image-captioning-base
+--blip_batch_size=4
+--num_captions=5
+--max_new_tokens=30
+--num_beams=5
+--clip_model_name=openai/clip-vit-base-patch32
+--clip_batch_size=8
+--clip_score_normalize=false
+--runner=DataflowRunner
+--experiments=use_runner_v2
diff --git
a/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Classification_Rightfit.txt
b/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Classification_Rightfit.txt
new file mode 100644
index 00000000000..1140ee5735e
--- /dev/null
+++
b/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Classification_Rightfit.txt
@@ -0,0 +1,41 @@
+# 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.
+
+--region=us-central1
+--machine_type=n1-standard-4
+--num_workers=50
+--disk_size_gb=50
+--autoscaling_algorithm=NONE
+--staging_location=gs://temp-storage-for-perf-tests/loadtests
+--temp_location=gs://temp-storage-for-perf-tests/loadtests
+--requirements_file=apache_beam/ml/inference/pytorch_rightfit_requirements.txt
+--publish_to_big_query=true
+--metrics_dataset=beam_run_inference
+--influx_measurement=torch_inference_imagenet_stream_rightfit
+--pretrained_model_name=efficientnet_b0
+--input_file=gs://apache-beam-ml/testing/inputs/openimage_50k_benchmark.txt
+--runner=DataflowRunner
+--mode=streaming
+--input_mode=gcs_uris
+--input_options={}
+--pubsub_topic=projects/apache-beam-testing/topics/images_topic
+--pubsub_subscription=projects/apache-beam-testing/subscriptions/images_subscription
+--model_state_dict_path=gs://apache-beam-ml/models/efficientnet_b0_state_dict.pth
+--image_size=224
+--top_k=5
+--inference_batch_size=16
+--window_sec=60
+--trigger_proc_time_sec=30
diff --git
a/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Object_Detection.txt
b/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Object_Detection.txt
new file mode 100644
index 00000000000..18f017ad010
--- /dev/null
+++
b/.github/workflows/load-tests-pipeline-options/beam_Inference_Python_Benchmarks_Dataflow_Pytorch_Image_Object_Detection.txt
@@ -0,0 +1,37 @@
+# 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.
+
+--region=us-central1
+--worker_machine_type=n1-standard-4
+--num_workers=50
+--disk_size_gb=50
+--autoscaling_algorithm=NONE
+--staging_location=gs://temp-storage-for-perf-tests/loadtests
+--temp_location=gs://temp-storage-for-perf-tests/loadtests
+--requirements_file=apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
+--publish_to_big_query=true
+--metrics_dataset=beam_run_inference
+--input_options={}
+--influx_measurement=result_torch_inference_image_object_detection_batch
+--pretrained_model_name=fasterrcnn_resnet50_fpn
+--inference_batch_size=8
+--resize_shorter_side=800
+--score_threshold=0.5
+--max_detections=50
+--input=gs://apache-beam-ml/testing/inputs/openimage_50k_benchmark.txt
+--model_state_dict_path=gs://apache-beam-ml/models/torchvision.detection.fasterrcnn_resnet50_fpn.pth
+--runner=DataflowRunner
+--experiments=use_runner_v2
diff --git a/.test-infra/tools/refresh_looker_metrics.py
b/.test-infra/tools/refresh_looker_metrics.py
index 69798e6f24e..351d8998f8f 100644
--- a/.test-infra/tools/refresh_looker_metrics.py
+++ b/.test-infra/tools/refresh_looker_metrics.py
@@ -43,6 +43,25 @@ LOOKS_TO_DOWNLOAD = [
("82", ["263", "264", "265", "266", "267"]), # PyTorch Sentiment
Streaming DistilBERT base uncased
("85", ["268", "269", "270", "271", "272"]), # PyTorch Sentiment Batch
DistilBERT base uncased
("86", ["284", "285", "286", "287", "288"]), # VLLM Batch Gemma
+
+ # PyTorch Image Classification EfficientNet-B0 Streaming (Right-fit) CPU
+ ("92", ["289", "290", "291", "292", "293"]),
+ ("97", ["306", "307", "308", "309", "310"]),
+ ("98", ["311", "312", "313", "314", "315"]),
+ ("99", ["316", "317", "318", "319", "320"]),
+
+ # PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch CPU
+ ("93", ["294", "295", "296", "298", "299"]),
+ ("100", ["321", "322", "323", "324", "325"]),
+ ("101", ["326", "327", "328", "329", "330"]),
+ ("102", ["331", "332", "333", "334", "335"]),
+
+ # PyTorch Image Captioning BLIP + CLIP Batch CPU
+ ("94", ["297", "300", "301", "302", "303"]),
+ ("103", ["336", "337", "338", "339", "340"]),
+ ("104", ["341", "342", "343", "344", "345"]),
+ ("105", ["346", "347", "348", "349", "350"]),
+
("96", ["270", "304", "305", "353", "354"]), # Table Row Inference
Sklearn Batch
("106", ["355", "356", "357", "358", "359"]), # Table Row Inference
Sklearn Streaming
("107", ["360", "361", "362", "363", "364"]), # MLTransform Generate
Vocab Batch
diff --git
a/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py
b/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py
new file mode 100644
index 00000000000..eb9fa903e61
--- /dev/null
+++ b/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py
@@ -0,0 +1,641 @@
+#
+# 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.
+#
+
+"""This pipeline performs image captioning using a multi-model approach:
+BLIP generates candidate captions, CLIP ranks them by image-text similarity.
+
+The pipeline reads image URIs from a GCS input file, decodes images, runs BLIP
+caption generation in batches on GPU, then runs CLIP ranking in batches on GPU.
+Results are written to BigQuery.
+"""
+
+import argparse
+import io
+import json
+import logging
+import threading
+import time
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Tuple
+
+import apache_beam as beam
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.ml.inference.base import KeyedModelHandler
+from apache_beam.ml.inference.base import ModelHandler
+from apache_beam.ml.inference.base import PredictionResult
+from apache_beam.ml.inference.base import RunInference
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.options.pipeline_options import SetupOptions
+from apache_beam.options.pipeline_options import StandardOptions
+from apache_beam.runners.runner import PipelineResult
+from apache_beam.transforms import window
+
+from google.api_core.exceptions import NotFound
+from google.cloud import pubsub_v1
+import torch
+import PIL.Image as PILImage
+
+# ============ Utility ============
+
+
+def now_millis() -> int:
+ return int(time.time() * 1000)
+
+
+def decode_pil(image_bytes: bytes) -> PILImage.Image:
+ with PILImage.open(io.BytesIO(image_bytes)) as img:
+ img = img.convert("RGB")
+ img.load()
+ return img
+
+
+# ============ DoFns ============
+
+
+class MakeKeyDoFn(beam.DoFn):
+ """Produce (uri, uri) so the URI is used as the stable key."""
+ def process(self, element: str):
+ uri = element
+ yield uri, uri
+
+
+class ReadImageBytesDoFn(beam.DoFn):
+ """Turn (uri, uri) -> (uri, dict(image_bytes))."""
+ def process(self, kv: Tuple[str, str]):
+ uri, _ = kv
+ try:
+ with FileSystems.open(uri) as f:
+ image_bytes = f.read()
+ yield uri, {"image_bytes": image_bytes}
+ except OSError as e:
+ logging.warning("Failed to read image %s: %s", uri, e)
+ return
+
+
+class DecodeImageDoFn(beam.DoFn):
+ """Turn (uri, dict(image_bytes)) -> (uri, dict(image))."""
+ def process(self, kv: Tuple[str, Dict[str, Any]]):
+ uri, value = kv
+ image_bytes = value["image_bytes"]
+
+ try:
+ image = decode_pil(image_bytes)
+ except (OSError, ValueError) as e:
+ logging.warning("Failed to decode image %s: %s", uri, e)
+ image = PILImage.new("RGB", (224, 224), color=(0, 0, 0))
+
+ yield uri, {"image": image}
+
+
+class PostProcessDoFn(beam.DoFn):
+ """Final PredictionResult -> row for BigQuery."""
+ def __init__(self, blip_name: str, clip_name: str):
+ self.blip_name = blip_name
+ self.clip_name = clip_name
+
+ def process(self, kv: Tuple[str, PredictionResult]):
+ uri, pred = kv
+ if hasattr(pred, "inference"):
+ inf = pred.inference or {}
+ else:
+ inf = pred
+ # Expected inference fields from CLIP handler:
+ # best_caption, best_score, candidates, scores, blip_ms, clip_ms, total_ms
+ best_caption = inf.get("best_caption", "")
+ best_score = inf.get("best_score", None)
+ candidates = inf.get("candidates", [])
+ scores = inf.get("scores", [])
+ blip_ms = inf.get("blip_ms", None)
+ clip_ms = inf.get("clip_ms", None)
+ total_ms = inf.get("total_ms", None)
+
+ yield {
+ "image_id": uri,
+ "blip_model": self.blip_name,
+ "clip_model": self.clip_name,
+ "best_caption": best_caption,
+ "best_score": float(best_score) if best_score is not None else None,
+ "candidates": json.dumps(candidates),
+ "scores": json.dumps(scores),
+ "blip_ms": int(blip_ms) if blip_ms is not None else None,
+ "clip_ms": int(clip_ms) if clip_ms is not None else None,
+ "total_ms": int(total_ms) if total_ms is not None else None,
+ "infer_ms": now_millis(),
+ }
+
+
+# ============ Model Handlers ============
+
+
+class BlipCaptionModelHandler(ModelHandler):
+ def __init__(
+ self,
+ model_name: str,
+ device: str,
+ batch_size: int,
+ num_captions: int,
+ max_new_tokens: int,
+ num_beams: int):
+ self.model_name = model_name
+ self.device = device
+ self.batch_size = batch_size
+ self.num_captions = num_captions
+ self.max_new_tokens = max_new_tokens
+ self.num_beams = num_beams
+
+ def load_model(self):
+ from transformers import BlipForConditionalGeneration, BlipProcessor
+ processor = BlipProcessor.from_pretrained(self.model_name)
+ model = BlipForConditionalGeneration.from_pretrained(self.model_name)
+ model.to(self.device)
+ model.eval()
+ return (model, processor)
+
+ def batch_elements_kwargs(self):
+ return {"max_batch_size": self.batch_size}
+
+ def run_inference(
+ self, batch: List[Dict[str, Any]], model_bundle, inference_args=None):
+
+ model, processor = model_bundle
+ start = now_millis()
+
+ images = [x["image"] for x in batch]
+
+ # Processor makes pixel_values
+ inputs = processor(images=images, return_tensors="pt")
+ pixel_values = inputs["pixel_values"].to(self.device)
+
+ # Generate captions
+ # We use num_return_sequences to generate multiple candidates per image.
+ # Note: this will produce (B * num_captions) sequences.
+ with torch.no_grad():
+ generated_ids = model.generate(
+ pixel_values=pixel_values,
+ max_new_tokens=self.max_new_tokens,
+ num_beams=max(self.num_beams, self.num_captions),
+ num_return_sequences=self.num_captions,
+ do_sample=False,
+ )
+
+ captions_all = processor.batch_decode(
+ generated_ids, skip_special_tokens=True)
+
+ # Group candidates per image
+ candidates_per_image = []
+ idx = 0
+ for _ in range(len(batch)):
+ candidates_per_image.append(captions_all[idx:idx + self.num_captions])
+ idx += self.num_captions
+
+ blip_ms = now_millis() - start
+
+ results = []
+ for i in range(len(batch)):
+ results.append({
+ "image": images[i],
+ "candidates": candidates_per_image[i],
+ "blip_ms": blip_ms,
+ })
+ return results
+
+ def get_metrics_namespace(self) -> str:
+ return "blip_captioning"
+
+
+class ClipRankModelHandler(ModelHandler):
+ def __init__(
+ self,
+ model_name: str,
+ device: str,
+ batch_size: int,
+ score_normalize: bool):
+ self.model_name = model_name
+ self.device = device
+ self.batch_size = batch_size
+ self.score_normalize = score_normalize
+
+ def load_model(self):
+ from transformers import CLIPModel, CLIPProcessor
+ processor = CLIPProcessor.from_pretrained(self.model_name)
+ model = CLIPModel.from_pretrained(self.model_name)
+ model.to(self.device)
+ model.eval()
+ return (model, processor)
+
+ def batch_elements_kwargs(self):
+ return {"max_batch_size": self.batch_size}
+
+ def run_inference(
+ self, batch: List[Dict[str, Any]], model_bundle, inference_args=None):
+
+ model, processor = model_bundle
+ start_batch = now_millis()
+
+ # Flat lists for a single batched CLIP forward pass
+ images: List[PILImage.Image] = []
+ texts: List[str] = []
+ offsets: List[Tuple[int, int, int]] = []
+ candidates_list: List[List[str]] = []
+ blip_ms_list: List[Optional[int]] = []
+
+ for x in batch:
+ img = x["image"]
+ candidates = [str(c) for c in (x.get("candidates", []) or [])]
+ candidates_list.append(candidates)
+ blip_ms_list.append(x.get("blip_ms", None))
+
+ image_idx = len(images)
+ images.append(img)
+
+ start_i = len(texts)
+ texts.extend(candidates)
+ end_i = len(texts)
+ offsets.append((image_idx, start_i, end_i))
+
+ results: List[Dict[str, Any]] = []
+
+ # Fast path: no candidates at all
+ if not texts:
+ for blip_ms in blip_ms_list:
+ total_ms = int(blip_ms) if blip_ms is not None else None
+ results.append({
+ "best_caption": "",
+ "best_score": None,
+ "candidates": [],
+ "scores": [],
+ "blip_ms": blip_ms,
+ "clip_ms": 0,
+ "total_ms": total_ms,
+ })
+ return results
+
+ with torch.no_grad():
+ image_inputs = processor(
+ images=images,
+ return_tensors="pt",
+ ).to(self.device)
+
+ text_inputs = processor(
+ text=texts,
+ return_tensors="pt",
+ padding=True,
+ truncation=True,
+ ).to(self.device)
+
+ image_features = model.get_image_features(
+ pixel_values=image_inputs["pixel_values"])
+ text_features = model.get_text_features(
+ input_ids=text_inputs["input_ids"],
+ attention_mask=text_inputs.get("attention_mask"),
+ )
+
+ image_features = image_features / image_features.norm(
+ dim=-1, keepdim=True)
+ text_features = text_features / text_features.norm(dim=-1, keepdim=True)
+
+ logit_scale = model.logit_scale.exp()
+
+ batch_ms = now_millis() - start_batch
+ total_pairs = len(texts)
+
+ items = zip(offsets, candidates_list, blip_ms_list)
+ for (image_idx, start_i, end_i), candidates, blip_ms in items:
+ if start_i == end_i:
+ total_ms = int(blip_ms) if blip_ms is not None else None
+ results.append({
+ "best_caption": "",
+ "best_score": None,
+ "candidates": [],
+ "scores": [],
+ "blip_ms": blip_ms,
+ "clip_ms": 0,
+ "total_ms": total_ms,
+ })
+ continue
+
+ candidate_features = text_features[start_i:end_i]
+ image_feature = image_features[image_idx].unsqueeze(0)
+
+ pair_scores = (candidate_features *
+ image_feature).sum(dim=-1) * logit_scale
+
+ scores = pair_scores.detach().cpu().tolist()
+
+ if self.score_normalize:
+ scores_t = torch.tensor(scores, dtype=torch.float32)
+ scores = torch.softmax(scores_t, dim=0).tolist()
+
+ best_idx = max(range(len(scores)), key=lambda i, s=scores: s[i])
+
+ pairs = end_i - start_i
+ clip_ms_elem = int(batch_ms * (pairs / max(1, total_pairs)))
+ if pairs > 0:
+ clip_ms_elem = max(1, clip_ms_elem)
+
+ total_ms = int(blip_ms) + clip_ms_elem if blip_ms is not None else None
+ results.append({
+ "best_caption": candidates[best_idx],
+ "best_score": float(scores[best_idx]),
+ "candidates": candidates,
+ "scores": scores,
+ "blip_ms": blip_ms,
+ "clip_ms": clip_ms_elem,
+ "total_ms": total_ms,
+ })
+
+ return results
+
+ def get_metrics_namespace(self) -> str:
+ return "clip_ranking"
+
+
+# ============ Args & Helpers ============
+
+
+def parse_known_args(argv):
+ parser = argparse.ArgumentParser()
+
+ # I/O & runtime
+ parser.add_argument(
+ '--mode', default='streaming', choices=['streaming', 'batch'])
+ parser.add_argument(
+ '--project', default='apache-beam-testing', help='GCP project ID')
+ parser.add_argument(
+ '--input', required=True, help='GCS path to file with image URIs')
+ parser.add_argument(
+ '--pubsub_topic',
+ default='projects/apache-beam-testing/topics/images_topic')
+ parser.add_argument(
+ '--pubsub_subscription',
+ default='projects/apache-beam-testing/subscriptions/images_subscription')
+ parser.add_argument(
+ '--output_table',
+ required=True,
+ help='BigQuery output table: dataset.table')
+ parser.add_argument(
+ '--publish_to_big_query', default='true', choices=['true', 'false'])
+ parser.add_argument(
+ '--feeder_start_delay_sec',
+ type=int,
+ default=900,
+ help=(
+ 'Delay before starting the feeder pipeline that reads URIs from GCS '
+ 'and publishes them to Pub/Sub. This delay allows the main streaming
'
+ 'pipeline workers to start and scale before data ingestion begins.'),
+ )
+
+ # Device
+ parser.add_argument('--device', default='GPU', choices=['CPU', 'GPU'])
+
+ # BLIP
+ parser.add_argument(
+ '--blip_model_name', default='Salesforce/blip-image-captioning-base')
+ parser.add_argument('--blip_batch_size', type=int, default=4)
+ parser.add_argument('--num_captions', type=int, default=5)
+ parser.add_argument('--max_new_tokens', type=int, default=30)
+ parser.add_argument('--num_beams', type=int, default=5)
+
+ # CLIP
+ parser.add_argument(
+ '--clip_model_name', default='openai/clip-vit-base-patch32')
+ parser.add_argument('--clip_batch_size', type=int, default=8)
+ parser.add_argument(
+ '--clip_score_normalize', default='false', choices=['true', 'false'])
+
+ # Windows
+ parser.add_argument('--window_sec', type=int, default=60)
+ parser.add_argument('--trigger_proc_time_sec', type=int, default=30)
+
+ known_args, pipeline_args = parser.parse_known_args(argv)
+ return known_args, pipeline_args
+
+
+def ensure_pubsub_resources(
+ project: str, topic_path: str, subscription_path: str):
+ publisher = pubsub_v1.PublisherClient()
+ subscriber = pubsub_v1.SubscriberClient()
+
+ try:
+ publisher.get_topic(request={"topic": topic_path})
+ except NotFound:
+ publisher.create_topic(name=topic_path)
+
+ try:
+ subscriber.get_subscription(request={"subscription": subscription_path})
+ except NotFound:
+ subscriber.create_subscription(name=subscription_path, topic=topic_path)
+
+
+def cleanup_pubsub_resources(
+ project: str, topic_path: str, subscription_path: str):
+ publisher = pubsub_v1.PublisherClient()
+ subscriber = pubsub_v1.SubscriberClient()
+
+ try:
+ subscriber.delete_subscription(request={"subscription": subscription_path})
+ logging.info(f"Deleted subscription: {subscription_path}")
+ except NotFound:
+ logging.info(f"Subscription already deleted: {subscription_path}")
+
+ try:
+ publisher.delete_topic(request={"topic": topic_path})
+ logging.info(f"Deleted topic: {topic_path}")
+ except NotFound:
+ logging.info(f"Topic already deleted: {topic_path}")
+
+ try:
+ subscriber.delete_subscription(
+ request={"subscription": full_subscription_path})
+ logging.info(f"Deleted subscription: {subscription_name}")
+ except NotFound:
+ logging.info(f"Subscription already deleted: {subscription_name}")
+
+ try:
+ publisher.delete_topic(request={"topic": full_topic_path})
+ logging.info(f"Deleted topic: {topic_name}")
+ except NotFound:
+ logging.info(f"Topic already deleted: {topic_name}")
+
+
+def override_or_add(args, flag, value):
+ if flag in args:
+ idx = args.index(flag)
+ args[idx + 1] = str(value)
+ else:
+ args.extend([flag, str(value)])
+
+
+# ============ Load pipeline ============
+
+
+def run_load_pipeline(known_args, pipeline_args):
+ """Reads GCS file with URIs and publishes them to Pub/Sub (for streaming)."""
+
+ pipeline_args = list(pipeline_args)
+ # enforce smaller/CPU-only defaults for feeder
+ override_or_add(pipeline_args, '--device', 'CPU')
+ override_or_add(pipeline_args, '--num_workers', '5')
+ override_or_add(pipeline_args, '--max_num_workers', '10')
+ override_or_add(
+ pipeline_args, '--job_name', f"images-load-pubsub-{int(time.time())}")
+ override_or_add(pipeline_args, '--project', known_args.project)
+ pipeline_args = [
+ arg for arg in pipeline_args if not arg.startswith("--experiments")
+ ]
+
+ pipeline_options = PipelineOptions(pipeline_args)
+ pipeline = beam.Pipeline(options=pipeline_options)
+
+ _ = (
+ pipeline
+ | 'ReadGCSFile' >> beam.io.ReadFromText(known_args.input)
+ | 'FilterEmpty' >> beam.Filter(lambda line: line.strip())
+ | 'ToBytes' >> beam.Map(lambda line: line.encode('utf-8'))
+ | 'ToPubSub' >> beam.io.WriteToPubSub(topic=known_args.pubsub_topic))
+ return pipeline.run()
+
+
+# ============ Main pipeline ============
+
+
+def run(
+ argv=None, save_main_session=True, test_pipeline=None) -> PipelineResult:
+ known_args, pipeline_args = parse_known_args(argv)
+
+ if known_args.mode == 'streaming':
+ ensure_pubsub_resources(
+ project=known_args.project,
+ topic_path=known_args.pubsub_topic,
+ subscription_path=known_args.pubsub_subscription)
+
+ # Start feeder thread that reads URIs from GCS and fills Pub/Sub.
+ # Delay is used to allow the main streaming pipeline workers to start
+ # and autoscale before the feeder pipeline begins publishing messages.
+ threading.Thread(
+ target=lambda: (
+ time.sleep(known_args.feeder_start_delay_sec), run_load_pipeline(
+ known_args, pipeline_args)),
+ daemon=True).start()
+
+ pipeline_options = PipelineOptions(pipeline_args)
+ pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
+ pipeline_options.view_as(StandardOptions).streaming = (
+ known_args.mode == 'streaming')
+
+ device = 'cuda' if known_args.device.upper() == 'GPU' else 'cpu'
+ clip_score_normalize = (known_args.clip_score_normalize == 'true')
+
+ blip_handler = BlipCaptionModelHandler(
+ model_name=known_args.blip_model_name,
+ device=device,
+ batch_size=int(known_args.blip_batch_size),
+ num_captions=int(known_args.num_captions),
+ max_new_tokens=int(known_args.max_new_tokens),
+ num_beams=int(known_args.num_beams),
+ )
+
+ clip_handler = ClipRankModelHandler(
+ model_name=known_args.clip_model_name,
+ device=device,
+ batch_size=int(known_args.clip_batch_size),
+ score_normalize=clip_score_normalize,
+ )
+
+ pipeline = test_pipeline or beam.Pipeline(options=pipeline_options)
+
+ if known_args.mode == 'batch':
+ pcoll = (
+ pipeline
+ | 'ReadURIsBatch' >> beam.io.ReadFromText(known_args.input)
+ | 'FilterEmptyBatch' >> beam.Filter(lambda s: s.strip()))
+ else:
+ pcoll = (
+ pipeline
+ | 'ReadFromPubSub' >>
+ beam.io.ReadFromPubSub(subscription=known_args.pubsub_subscription)
+ | 'DecodeUTF8' >> beam.Map(lambda x: x.decode('utf-8'))
+ | 'Window' >> beam.WindowInto(
+ window.FixedWindows(known_args.window_sec),
+ trigger=beam.trigger.AfterProcessingTime(
+ known_args.trigger_proc_time_sec),
+ accumulation_mode=beam.trigger.AccumulationMode.DISCARDING,
+ allowed_lateness=0))
+
+ keyed = (pcoll | 'MakeKey' >> beam.ParDo(MakeKeyDoFn()))
+ image_bytes = (keyed | 'ReadImageBytes' >> beam.ParDo(ReadImageBytesDoFn()))
+ images = (image_bytes | 'DecodeImage' >> beam.ParDo(DecodeImageDoFn()))
+
+ # Stage 1: BLIP candidate generation
+ blip_out = (
+ images
+ | 'RunInferenceBLIP' >> RunInference(KeyedModelHandler(blip_handler)))
+
+ # Stage 2: CLIP ranking over candidates
+ clip_out = (
+ blip_out
+ | 'RunInferenceCLIP' >> RunInference(KeyedModelHandler(clip_handler)))
+
+ results = (
+ clip_out
+ | 'PostProcess' >> beam.ParDo(
+ PostProcessDoFn(
+ blip_name=known_args.blip_model_name,
+ clip_name=known_args.clip_model_name)))
+
+ method = (
+ beam.io.WriteToBigQuery.Method.FILE_LOADS if known_args.mode == 'batch'
+ else beam.io.WriteToBigQuery.Method.STREAMING_INSERTS)
+
+ if known_args.publish_to_big_query == 'true':
+ _ = (
+ results
+ | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
+ known_args.output_table,
+ schema=(
+ 'image_id:STRING, blip_model:STRING, clip_model:STRING, '
+ 'best_caption:STRING, best_score:FLOAT, '
+ 'candidates:STRING, scores:STRING, '
+ 'blip_ms:INT64, clip_ms:INT64, total_ms:INT64,
infer_ms:INT64'),
+ write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
+ create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
+ method=method))
+
+ result = pipeline.run()
+ try:
+ result.wait_until_finish(duration=1800000) # 30 min
+ finally:
+ try:
+ result.cancel()
+ except Exception:
+ logging.debug("Failed to cancel pipeline result.", exc_info=True)
+
+ if known_args.mode == 'streaming':
+ cleanup_pubsub_resources(
+ project=known_args.project,
+ topic_path=known_args.pubsub_topic,
+ subscription_path=known_args.pubsub_subscription)
+
+ return result
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ run()
diff --git
a/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py
b/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py
new file mode 100644
index 00000000000..a8eef5a7d5e
--- /dev/null
+++
b/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py
@@ -0,0 +1,517 @@
+#
+# 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.
+#
+
+"""This pipeline performs object detection using an open-source PyTorch
+TorchVision detection model (e.g., Faster R-CNN ResNet50 FPN) on GPU.
+
+It reads image URIs from a GCS input file, decodes and preprocesses images,
+runs batched GPU inference via RunInference, post-processes detection outputs,
+and writes results to BigQuery.
+
+The pipeline targets stable and reproducible performance measurements for
+GPU inference workloads (no right-fitting; fixed batch size).
+"""
+
+import argparse
+import io
+import json
+import logging
+import threading
+import time
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Sequence
+from typing import Tuple
+
+import apache_beam as beam
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.ml.inference.base import KeyedModelHandler
+from apache_beam.ml.inference.base import PredictionResult
+from apache_beam.ml.inference.base import RunInference
+from apache_beam.ml.inference.pytorch_inference import
PytorchModelHandlerTensor
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.options.pipeline_options import SetupOptions
+from apache_beam.options.pipeline_options import StandardOptions
+from apache_beam.runners.runner import PipelineResult
+from apache_beam.transforms import window
+
+from google.api_core.exceptions import NotFound
+from google.cloud import pubsub_v1
+import torch
+import PIL.Image as PILImage
+
+# ============ Utility & Preprocessing ============
+
+
+def now_millis() -> int:
+ return int(time.time() * 1000)
+
+
+def decode_to_tens(image_bytes: bytes, image_size: int = 800) -> torch.Tensor:
+ """Decode bytes -> RGB PIL -> resize/pad square -> float tensor [0..1], CHW.
+
+ TorchVision detection models accept float tensors in [0..1]. We force a fixed
+ square shape so PytorchModelHandlerTensor can batch tensors with torch.stack.
+ """
+ with PILImage.open(io.BytesIO(image_bytes)) as img:
+ img = img.convert("RGB")
+
+ w, h = img.size
+ scale = min(image_size / float(w), image_size / float(h))
+ new_w = max(1, int(round(w * scale)))
+ new_h = max(1, int(round(h * scale)))
+
+ img = img.resize((new_w, new_h))
+
+ padded = PILImage.new("RGB", (image_size, image_size), color=(0, 0, 0))
+ left = (image_size - new_w) // 2
+ top = (image_size - new_h) // 2
+ padded.paste(img, (left, top))
+
+ import numpy as np
+ arr = np.asarray(padded).astype("float32") / 255.0
+ arr = np.transpose(arr, (2, 0, 1))
+ return torch.from_numpy(arr).float()
+
+
+# ============ DoFns ============
+
+
+class MakeKeyDoFn(beam.DoFn):
+ """Produce (uri, uri) where the URI is used as the stable key."""
+ def process(self, element: str):
+ uri = element
+ yield uri, uri
+
+
+class DecodePreprocessDoFn(beam.DoFn):
+ """Turn (uri, uri) -> (uri, tensor)."""
+ def __init__(self, image_size: int = 800):
+ self.image_size = image_size
+
+ def process(self, kv: Tuple[str, str]):
+ uri, _ = kv
+ start = now_millis()
+ try:
+ with FileSystems.open(uri) as f:
+ image_bytes = f.read()
+ tensor = decode_to_tens(image_bytes, image_size=self.image_size)
+ preprocess_ms = now_millis() - start
+ yield uri, {"tensor": tensor, "preprocess_ms": preprocess_ms}
+ except (OSError, ValueError):
+ logging.exception("Decode failed for %s", uri)
+ return
+
+
+# pylint: disable=unused-argument
+def _torchvision_detection_inference_fn(
+ batch: Sequence[torch.Tensor],
+ model: torch.nn.Module,
+ device: torch.device,
+ inference_args: Optional[dict[str, Any]] = None,
+ model_id: Optional[str] = None,
+) -> List[Dict[str, Any]]:
+ """Inference function for TorchVision detection models.
+
+ TorchVision detection models expect List[Tensor] where each tensor is:
+ - shape: [3, H, W]
+ - dtype: float32
+ - values: [0..1]
+ """
+
+ with torch.no_grad():
+ # Ensure tensors are on device
+ inputs = []
+ for t in batch:
+ if isinstance(t, torch.Tensor):
+ inputs.append(t.to(device))
+ else:
+ # Defensive: if somehow non-tensor slips through.
+ inputs.append(torch.as_tensor(t).to(device))
+ outputs = model(inputs) # List[Dict[str, Tensor]]
+ return outputs
+
+
+class PostProcessDoFn(beam.DoFn):
+ """PredictionResult -> dict row for BQ."""
+ def __init__(
+ self, model_name: str, score_threshold: float, max_detections: int):
+ self.model_name = model_name
+ self.score_threshold = score_threshold
+ self.max_detections = max_detections
+
+ def _extract_detection(self, inference_obj: Any) -> Dict[str, Any]:
+ """Extract detection fields from torchvision output dict."""
+ # Expect: {'boxes': Tensor[N,4], 'labels': Tensor[N], 'scores': Tensor[N]}
+ boxes = inference_obj.get("boxes")
+ labels = inference_obj.get("labels")
+ scores = inference_obj.get("scores")
+
+ # Convert to CPU lists
+ if isinstance(scores, torch.Tensor):
+ scores_list = scores.detach().cpu().tolist()
+ else:
+ scores_list = list(scores) if scores is not None else []
+
+ if isinstance(labels, torch.Tensor):
+ labels_list = labels.detach().cpu().tolist()
+ else:
+ labels_list = list(labels) if labels is not None else []
+
+ if isinstance(boxes, torch.Tensor):
+ boxes_list = boxes.detach().cpu().tolist()
+ else:
+ boxes_list = list(boxes) if boxes is not None else []
+
+ # Filter by threshold and trim to max_detections
+ dets = []
+ for i in range(min(len(scores_list), len(labels_list), len(boxes_list))):
+ score = float(scores_list[i])
+ if score < self.score_threshold:
+ continue
+ box = boxes_list[i] # [x1,y1,x2,y2]
+ dets.append({
+ "label_id": int(labels_list[i]),
+ "score": score,
+ "box": [float(box[0]), float(box[1]), float(box[2]), float(box[3])],
+ })
+ if len(dets) >= self.max_detections:
+ break
+
+ return {
+ "detections": dets,
+ "num_detections": len(dets),
+ }
+
+ def process(self, kv: Tuple[str, PredictionResult]):
+ image_uri, pred = kv
+
+ # pred can be PredictionResult OR raw torchvision dict.
+ if hasattr(pred, "inference"):
+ inference_obj = pred.inference
+ else:
+ inference_obj = pred
+
+ if isinstance(inference_obj, list) and len(inference_obj) == 1:
+ inference_obj = inference_obj[0]
+
+ if not isinstance(inference_obj, dict):
+ logging.warning(
+ "Unexpected inf-ce type for %s: %s", image_uri, type(inference_obj))
+ yield {
+ "image_id": image_uri,
+ "model_name": self.model_name,
+ "detections": json.dumps([]),
+ "num_detections": 0,
+ "infer_ms": now_millis(),
+ }
+ return
+
+ extracted = self._extract_detection(inference_obj)
+
+ yield {
+ "image_id": image_uri,
+ "model_name": self.model_name,
+ "detections": json.dumps(extracted["detections"]),
+ "num_detections": int(extracted["num_detections"]),
+ "infer_ms": now_millis(),
+ }
+
+
+# ============ Args & Helpers ============
+
+
+def parse_known_args(argv):
+ parser = argparse.ArgumentParser()
+
+ # I/O & runtime
+ parser.add_argument(
+ '--project', default='apache-beam-testing', help='GCP project ID')
+ parser.add_argument(
+ '--mode', default='streaming', choices=['streaming', 'batch'])
+ parser.add_argument(
+ '--output_table',
+ required=True,
+ help='BigQuery output table: dataset.table')
+ parser.add_argument(
+ '--publish_to_big_query', default='true', choices=['true', 'false'])
+ parser.add_argument(
+ '--input', required=True, help='GCS path to file with image URIs')
+ parser.add_argument(
+ '--pubsub_topic',
+ default='projects/apache-beam-testing/topics/images_topic')
+ parser.add_argument(
+ '--pubsub_subscription',
+ default='projects/apache-beam-testing/subscriptions/images_subscription')
+ parser.add_argument(
+ '--feeder_start_delay_sec',
+ type=int,
+ default=900,
+ help=(
+ 'Delay before starting the feeder pipeline that reads URIs from GCS '
+ 'and publishes them to Pub/Sub. This delay allows the main streaming
'
+ 'pipeline workers to start and scale before data ingestion begins.'),
+ )
+
+ # Model & inference
+ parser.add_argument(
+ '--pretrained_model_name',
+ default='fasterrcnn_resnet50_fpn',
+ help=(
+ 'TorchVision detection model name '
+ '(e.g., fasterrcnn_resnet50_fpn)'))
+ parser.add_argument(
+ '--model_state_dict_path',
+ required=True,
+ help='GCS path to a state_dict .pth for the chosen model')
+ parser.add_argument('--device', default='GPU', choices=['CPU', 'GPU'])
+
+ # Batch sizing (no right-fitting)
+ parser.add_argument('--inference_batch_size', type=int, default=8)
+
+ # Preprocess
+ parser.add_argument('--image_size', type=int, default=800)
+
+ # Postprocess
+ parser.add_argument('--score_threshold', type=float, default=0.5)
+ parser.add_argument('--max_detections', type=int, default=50)
+
+ # Windows
+ parser.add_argument('--window_sec', type=int, default=60)
+ parser.add_argument('--trigger_proc_time_sec', type=int, default=30)
+
+ known_args, pipeline_args = parser.parse_known_args(argv)
+ return known_args, pipeline_args
+
+
+def ensure_pubsub_resources(
+ project: str, topic_path: str, subscription_path: str):
+ publisher = pubsub_v1.PublisherClient()
+ subscriber = pubsub_v1.SubscriberClient()
+
+ try:
+ publisher.get_topic(request={"topic": topic_path})
+ except NotFound:
+ publisher.create_topic(name=topic_path)
+
+ try:
+ subscriber.get_subscription(request={"subscription": subscription_path})
+ except NotFound:
+ subscriber.create_subscription(name=subscription_path, topic=topic_path)
+
+
+def cleanup_pubsub_resources(
+ project: str, topic_path: str, subscription_path: str):
+ publisher = pubsub_v1.PublisherClient()
+ subscriber = pubsub_v1.SubscriberClient()
+
+ try:
+ subscriber.delete_subscription(request={"subscription": subscription_path})
+ logging.info(f"Deleted subscription: {subscription_path}")
+ except NotFound:
+ logging.info(f"Subscription already deleted: {subscription_path}")
+
+ try:
+ publisher.delete_topic(request={"topic": topic_path})
+ logging.info(f"Deleted topic: {topic_path}")
+ except NotFound:
+ logging.info(f"Topic already deleted: {topic_path}")
+
+
+def override_or_add(args, flag, value):
+ if flag in args:
+ idx = args.index(flag)
+ args[idx + 1] = str(value)
+ else:
+ args.extend([flag, str(value)])
+
+
+def create_torchvision_detection_model(model_name: str):
+ """Creates a TorchVision detection model instance.
+
+ Note: We will load weights via state_dict_path (required by Beam handler when
+ model_class is provided).
+ """
+ import torchvision
+
+ name = model_name.strip()
+
+ if name == "fasterrcnn_resnet50_fpn":
+ model = torchvision.models.detection.fasterrcnn_resnet50_fpn(weights=None)
+ elif name == "retinanet_resnet50_fpn":
+ model = torchvision.models.detection.retinanet_resnet50_fpn(weights=None)
+ else:
+ raise ValueError(f"Unsupported detection model: {model_name}")
+
+ model.eval()
+ return model
+
+
+# ============ Load pipeline ============
+
+
+def run_load_pipeline(known_args, pipeline_args):
+ """Reads GCS file with URIs and publishes them to Pub/Sub (for streaming)."""
+
+ pipeline_args = list(pipeline_args)
+ # enforce smaller/CPU-only defaults for feeder
+ override_or_add(pipeline_args, '--device', 'CPU')
+ override_or_add(pipeline_args, '--num_workers', '5')
+ override_or_add(pipeline_args, '--max_num_workers', '10')
+ override_or_add(
+ pipeline_args, '--job_name', f"images-load-pubsub-{int(time.time())}")
+ override_or_add(pipeline_args, '--project', known_args.project)
+ pipeline_args = [
+ arg for arg in pipeline_args if not arg.startswith("--experiments")
+ ]
+
+ pipeline_options = PipelineOptions(pipeline_args)
+ pipeline = beam.Pipeline(options=pipeline_options)
+
+ _ = (
+ pipeline
+ | 'ReadGCSFile' >> beam.io.ReadFromText(known_args.input)
+ | 'FilterEmpty' >> beam.Filter(lambda line: line.strip())
+ | 'ToBytes' >> beam.Map(lambda line: line.encode('utf-8'))
+ | 'ToPubSub' >> beam.io.WriteToPubSub(topic=known_args.pubsub_topic))
+ return pipeline.run()
+
+
+# ============ Main pipeline ============
+
+
+def run(
+ argv=None, save_main_session=True, test_pipeline=None) -> PipelineResult:
+ known_args, pipeline_args = parse_known_args(argv)
+
+ if known_args.mode == 'streaming':
+ ensure_pubsub_resources(
+ project=known_args.project,
+ topic_path=known_args.pubsub_topic,
+ subscription_path=known_args.pubsub_subscription)
+
+ # Start feeder thread that reads URIs from GCS and fills Pub/Sub.
+ # Delay is used to allow the main streaming pipeline workers to start
+ # and autoscale before the feeder pipeline begins publishing messages.
+ threading.Thread(
+ target=lambda: (
+ time.sleep(known_args.feeder_start_delay_sec), run_load_pipeline(
+ known_args, pipeline_args)),
+ daemon=True).start()
+
+ pipeline_options = PipelineOptions(pipeline_args)
+ pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
+ pipeline_options.view_as(StandardOptions).streaming = (
+ known_args.mode == 'streaming')
+
+ device = 'GPU' if known_args.device.upper() == 'GPU' else 'CPU'
+
+ # Fixed batch size (no right-fitting)
+ batch_size = int(known_args.inference_batch_size)
+
+ model_handler = PytorchModelHandlerTensor(
+ model_class=lambda: create_torchvision_detection_model(
+ known_args.pretrained_model_name),
+ model_params={},
+ state_dict_path=known_args.model_state_dict_path,
+ device=device,
+ inference_batch_size=batch_size,
+ inference_fn=_torchvision_detection_inference_fn,
+ )
+
+ pipeline = test_pipeline or beam.Pipeline(options=pipeline_options)
+
+ if known_args.mode == 'batch':
+ pcoll = (
+ pipeline
+ | 'ReadURIsBatch' >> beam.io.ReadFromText(known_args.input)
+ | 'FilterEmptyBatch' >> beam.Filter(lambda s: s.strip()))
+ else:
+ pcoll = (
+ pipeline
+ | 'ReadFromPubSub' >>
+ beam.io.ReadFromPubSub(subscription=known_args.pubsub_subscription)
+ | 'DecodeUTF8' >> beam.Map(lambda x: x.decode('utf-8'))
+ | 'Window' >> beam.WindowInto(
+ window.FixedWindows(known_args.window_sec),
+ trigger=beam.trigger.AfterProcessingTime(
+ known_args.trigger_proc_time_sec),
+ accumulation_mode=beam.trigger.AccumulationMode.DISCARDING,
+ allowed_lateness=0))
+
+ keyed = (pcoll | 'MakeKey' >> beam.ParDo(MakeKeyDoFn()))
+
+ preprocessed = (
+ keyed
+ | 'DecodePreprocess' >> beam.ParDo(
+ DecodePreprocessDoFn(image_size=known_args.image_size)))
+
+ to_infer = (
+ preprocessed
+ | 'ToKeyedTensor' >> beam.Map(lambda kv: (kv[0], kv[1]["tensor"])))
+
+ predictions = (
+ to_infer
+ | 'RunInference' >> RunInference(KeyedModelHandler(model_handler)))
+
+ results = (
+ predictions
+ | 'PostProcess' >> beam.ParDo(
+ PostProcessDoFn(
+ model_name=known_args.pretrained_model_name,
+ score_threshold=known_args.score_threshold,
+ max_detections=known_args.max_detections)))
+
+ method = (
+ beam.io.WriteToBigQuery.Method.FILE_LOADS if known_args.mode == 'batch'
+ else beam.io.WriteToBigQuery.Method.STREAMING_INSERTS)
+
+ if known_args.publish_to_big_query == 'true':
+ _ = (
+ results
+ | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
+ known_args.output_table,
+ schema=(
+ 'image_id:STRING, model_name:STRING, '
+ 'detections:STRING, num_detections:INT64, infer_ms:INT64'),
+ write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
+ create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
+ method=method))
+
+ result = pipeline.run()
+ try:
+ result.wait_until_finish(duration=1800000) # 30 min
+ finally:
+ try:
+ result.cancel()
+ except Exception:
+ logging.debug("Failed to cancel pipeline result.", exc_info=True)
+
+ if known_args.mode == 'streaming':
+ cleanup_pubsub_resources(
+ project=known_args.project,
+ topic_path=known_args.pubsub_topic,
+ subscription_path=known_args.pubsub_subscription)
+
+ return result
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ run()
diff --git
a/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py
b/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py
new file mode 100644
index 00000000000..168303cb61c
--- /dev/null
+++ b/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py
@@ -0,0 +1,537 @@
+# 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.
+
+"""This pipeline performs image classification using an open-source
+PyTorch EfficientNet-B0 model optimized for T4 GPUs.
+It reads image URIs from Pub/Sub, decodes and preprocesses them in parallel,
+and runs inference with adaptive batch sizing for optimal GPU utilization.
+The pipeline targets stable and reproducible performance measurements under
+continuous load.
+Resources like Pub/Sub topic/subscription cleanup is handled programmatically.
+"""
+
+import argparse
+import io
+import json
+import logging
+import threading
+import time
+from typing import Optional
+from typing import Tuple
+
+import torch
+import torch.nn.functional as F
+
+import apache_beam as beam
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.ml.inference.base import KeyedModelHandler
+from apache_beam.ml.inference.base import PredictionResult
+from apache_beam.ml.inference.base import RunInference
+from apache_beam.ml.inference.pytorch_inference import
PytorchModelHandlerTensor
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.options.pipeline_options import SetupOptions
+from apache_beam.options.pipeline_options import StandardOptions
+from apache_beam.runners.runner import PipelineResult
+from apache_beam.transforms import window
+
+from google.api_core.exceptions import NotFound
+from google.cloud import pubsub_v1
+import PIL.Image as PILImage
+
+# ============ Utility & Preprocessing ============
+
+IMAGENET_MEAN = [0.485, 0.456, 0.406]
+IMAGENET_STD = [0.229, 0.224, 0.225]
+
+
+def now_millis() -> int:
+ return int(time.time() * 1000)
+
+
+def load_image_from_uri(uri: str) -> bytes:
+ with FileSystems.open(uri) as f:
+ return f.read()
+
+
+def decode_and_preprocess(image_bytes: bytes, size: int = 224) -> torch.Tensor:
+ """Decode bytes->RGB PIL->resize shorter side->center crop->normalize."""
+ with PILImage.open(io.BytesIO(image_bytes)) as img:
+ img = img.convert("RGB")
+
+ resize_size = 256
+ w, h = img.size
+ if w < h:
+ new_w = resize_size
+ new_h = int(h * resize_size / w)
+ else:
+ new_h = resize_size
+ new_w = int(w * resize_size / h)
+
+ img = img.resize((new_w, new_h))
+
+ w, h = img.size
+ left = (w - size) // 2
+ top = (h - size) // 2
+ img = img.crop((left, top, left + size, top + size))
+
+ import numpy as np
+ mean = np.array(IMAGENET_MEAN, dtype=np.float32)
+ std = np.array(IMAGENET_STD, dtype=np.float32)
+
+ arr = np.asarray(img).astype("float32") / 255.0
+ arr = (arr - mean) / std
+ arr = np.transpose(arr, (2, 0, 1)).astype("float32")
+ return torch.from_numpy(arr).float()
+
+
+class MakeKeyDoFn(beam.DoFn):
+ """Produce (image_id, payload) stable for dedup & BQ insertId."""
+ def __init__(self, input_mode: str):
+ self.input_mode = input_mode
+
+ def process(self, element: str | bytes):
+ # Input can be raw bytes from Pub/Sub or a GCS URI string, depends on mode
+ if self.input_mode == "bytes":
+ # element is bytes message, assume it includes
+ # {"image_id": "...", "bytes": base64?} or just raw bytes.
+ import hashlib
+ b = element if isinstance(element,
+ (bytes,
+ bytearray)) else element.encode('utf-8')
+ image_id = hashlib.sha1(b).hexdigest()
+ yield image_id, b
+ else:
+ # gcs_uris: element is uri string; image_id = sha1(uri)
+ import hashlib
+ uri = element.decode("utf-8") if isinstance(
+ element, (bytes, bytearray)) else str(element)
+ image_id = hashlib.sha1(uri.encode("utf-8")).hexdigest()
+ yield image_id, uri
+
+
+class DecodePreprocessDoFn(beam.DoFn):
+ """Turn (image_id, bytes|uri) -> (image_id, torch.Tensor)"""
+ def __init__(self, input_mode: str, image_size: int = 224):
+ self.input_mode = input_mode
+ self.image_size = image_size
+
+ def process(self, kv: Tuple[str, object]):
+ image_id, payload = kv
+ start = now_millis()
+
+ try:
+ if self.input_mode == "bytes":
+ b = payload if isinstance(payload,
+ (bytes, bytearray)) else bytes(payload)
+ else:
+ uri = payload if isinstance(payload, str) else payload.decode("utf-8")
+ b = load_image_from_uri(uri)
+
+ tensor = decode_and_preprocess(b, self.image_size)
+ preprocess_ms = now_millis() - start
+ yield image_id, {"tensor": tensor, "preprocess_ms": preprocess_ms}
+ except Exception as e:
+ logging.warning("Decode failed for %s: %s", image_id, e)
+ return
+
+
+class PostProcessDoFn(beam.DoFn):
+ """PredictionResult -> dict row for BQ."""
+ def __init__(self, top_k: int, model_name: str):
+ self.top_k = top_k
+ self.model_name = model_name
+
+ def process(self, kv: Tuple[str, PredictionResult]):
+ image_id, pred = kv
+
+ # pred can be PredictionResult OR raw inference object.
+ inference_obj = pred.inference if hasattr(pred, "inference") else pred
+
+ # inference_obj can be dict {'logits': tensor} OR tensor directly.
+ if isinstance(inference_obj, dict):
+ logits = inference_obj.get("logits", None)
+ if logits is None:
+ raise ValueError(
+ f"Unable to find 'logits' in model output. "
+ f"Available keys: {list(inference_obj.keys())}")
+ else:
+ logits = inference_obj
+
+ if not isinstance(logits, torch.Tensor):
+ logging.warning(
+ "Unexpected logits type for %s: %s", image_id, type(logits))
+ return
+
+ # Ensure shape [1, C]
+ if logits.ndim == 1:
+ logits = logits.unsqueeze(0)
+
+ probs = F.softmax(logits, dim=-1) # [B, C]
+ values, indices = torch.topk(
+ probs, k=min(self.top_k, probs.shape[-1]), dim=-1
+ )
+
+ topk = [{
+ "class_id": int(idx.item()), "score": float(val.item())
+ } for idx, val in zip(indices[0], values[0])]
+
+ yield {
+ "image_id": image_id,
+ "model_name": self.model_name,
+ "topk": json.dumps(topk),
+ "infer_ms": now_millis(),
+ }
+
+
+# ============ Args & Helpers ============
+
+
+def parse_known_args(argv):
+ parser = argparse.ArgumentParser()
+ # I/O & runtime
+ parser.add_argument(
+ '--project', default='apache-beam-testing', help='GCP project ID')
+ parser.add_argument(
+ '--mode', default='streaming', choices=['streaming', 'batch'])
+ parser.add_argument(
+ '--output_table',
+ required=True,
+ help='BigQuery output table: dataset.table')
+ parser.add_argument(
+ '--publish_to_big_query', default='true', choices=['true', 'false'])
+ parser.add_argument(
+ '--input_mode', default='gcs_uris', choices=['gcs_uris', 'bytes'])
+ parser.add_argument(
+ '--input',
+ required=True,
+ help='GCS path to file with URIs (for load) OR unused for bytes')
+ parser.add_argument(
+ '--pubsub_topic',
+ default='projects/apache-beam-testing/topics/images_topic')
+ parser.add_argument(
+ '--pubsub_subscription',
+ default='projects/apache-beam-testing/subscriptions/images_subscription')
+ parser.add_argument(
+ '--feeder_start_delay_sec',
+ type=int,
+ default=900,
+ help=(
+ 'Delay before starting the feeder pipeline that reads URIs from GCS '
+ 'and publishes them to Pub/Sub. This delay allows the main streaming
'
+ 'pipeline workers to start and scale before data ingestion begins.'),
+ )
+
+ # Model & inference
+ parser.add_argument(
+ '--pretrained_model_name',
+ default='efficientnet_b0',
+ help='OSS model name (e.g., efficientnet_b0|mobilenetv3_large_100)')
+ parser.add_argument(
+ '--model_state_dict_path',
+ default=None,
+ help='Optional state_dict to load')
+ parser.add_argument('--device', default='GPU', choices=['CPU', 'GPU'])
+ parser.add_argument('--image_size', type=int, default=224)
+ parser.add_argument('--top_k', type=int, default=5)
+ parser.add_argument(
+ '--inference_batch_size',
+ default='16',
+ help='int or "auto"; auto tries 64→32→16')
+
+ # Windows
+ parser.add_argument('--window_sec', type=int, default=60)
+ parser.add_argument('--trigger_proc_time_sec', type=int, default=30)
+
+ known_args, pipeline_args = parser.parse_known_args(argv)
+ return known_args, pipeline_args
+
+
+def ensure_pubsub_resources(
+ project: str, topic_path: str, subscription_path: str):
+ publisher = pubsub_v1.PublisherClient()
+ subscriber = pubsub_v1.SubscriberClient()
+
+ try:
+ publisher.get_topic(request={"topic": topic_path})
+ except NotFound:
+ publisher.create_topic(name=topic_path)
+
+ try:
+ subscriber.get_subscription(request={"subscription": subscription_path})
+ except NotFound:
+ subscriber.create_subscription(name=subscription_path, topic=topic_path)
+
+
+def cleanup_pubsub_resources(
+ project: str, topic_path: str, subscription_path: str):
+ publisher = pubsub_v1.PublisherClient()
+ subscriber = pubsub_v1.SubscriberClient()
+
+ try:
+ subscriber.delete_subscription(request={"subscription": subscription_path})
+ logging.info(f"Deleted subscription: {subscription_path}")
+ except NotFound:
+ logging.info(f"Subscription already deleted: {subscription_path}")
+
+ try:
+ publisher.delete_topic(request={"topic": topic_path})
+ logging.info(f"Deleted topic: {topic_path}")
+ except NotFound:
+ logging.info(f"Topic already deleted: {topic_path}")
+
+
+def override_or_add(args, flag, value):
+ if flag in args:
+ idx = args.index(flag)
+ args[idx + 1] = str(value)
+ else:
+ args.extend([flag, str(value)])
+
+
+# ============ Model factory (timm) ============
+
+
+def create_timm_m(model_name: str, num_classes: int = 1000):
+ import timm
+ model = timm.create_model(
+ model_name, pretrained=True, num_classes=num_classes)
+ model.eval()
+ return model
+
+
+def pick_batch_size(arg: str) -> Optional[int]:
+ if isinstance(arg, str) and arg.lower() == 'auto':
+ return None
+ try:
+ return int(arg)
+ except Exception:
+ return None
+
+
+class RightFittingPytorchModelHandlerTensor(PytorchModelHandlerTensor):
+ def __init__(self, batch_sizes_to_try, image_size, *args, **kwargs):
+ self._batch_sizes_to_try = batch_sizes_to_try
+ self._rightfit_image_size = image_size
+ self._selected_batch_size = batch_sizes_to_try[0]
+ super().__init__(*args, **kwargs)
+
+ def load_model(self):
+ model = super().load_model()
+ last_err = None
+
+ for bs in self._batch_sizes_to_try:
+ try:
+ model_device = next(model.parameters()).device
+ dummy = torch.zeros(
+ (bs, 3, self._rightfit_image_size, self._rightfit_image_size),
+ dtype=torch.float32,
+ device=model_device)
+
+ with torch.no_grad():
+ model(dummy)
+
+ self._selected_batch_size = bs
+ logging.info("Selected inference batch size: %s", bs)
+ return model
+ except RuntimeError as e:
+ last_err = e
+ logging.warning("Batch size %s failed during worker warmup: %s", bs, e)
+
+ del dummy
+ if torch.cuda.is_available():
+ torch.cuda.empty_cache()
+
+ raise RuntimeError(
+ f"No valid inference batch size found from {self._batch_sizes_to_try}"
+ ) from last_err
+
+ def run_inference(self, batch, model, inference_args=None):
+ results = []
+
+ for i in range(0, len(batch), self._selected_batch_size):
+ sub_batch = batch[i:i + self._selected_batch_size]
+ results.extend(super().run_inference(
+ sub_batch,
+ model,
+ inference_args,
+ ))
+
+ return results
+
+
+# ============ Load pipeline ============
+
+
+def run_load_pipeline(known_args, pipeline_args):
+ """Reads GCS file with URIs and publishes them to Pub/Sub (for streaming)."""
+
+ pipeline_args = list(pipeline_args)
+ # enforce smaller/CPU-only defaults for feeder
+ override_or_add(pipeline_args, '--device', 'CPU')
+ override_or_add(pipeline_args, '--num_workers', '5')
+ override_or_add(pipeline_args, '--max_num_workers', '10')
+ override_or_add(
+ pipeline_args, '--job_name', f"images-load-pubsub-{int(time.time())}")
+ override_or_add(pipeline_args, '--project', known_args.project)
+ pipeline_args = [
+ arg for arg in pipeline_args if not arg.startswith("--experiments")
+ ]
+
+ pipeline_options = PipelineOptions(pipeline_args)
+ pipeline = beam.Pipeline(options=pipeline_options)
+
+ _ = (
+ pipeline
+ | 'ReadGCSFile' >> beam.io.ReadFromText(known_args.input)
+ | 'FilterEmpty' >> beam.Filter(lambda line: line.strip())
+ | 'ToBytes' >> beam.Map(lambda line: line.encode('utf-8'))
+ | 'ToPubSub' >> beam.io.WriteToPubSub(topic=known_args.pubsub_topic))
+ return pipeline.run()
+
+
+# ============ Main pipeline ============
+
+
+def run(
+ argv=None, save_main_session=True, test_pipeline=None) -> PipelineResult:
+ known_args, pipeline_args = parse_known_args(argv)
+
+ if known_args.mode == 'streaming':
+ ensure_pubsub_resources(
+ project=known_args.project,
+ topic_path=known_args.pubsub_topic,
+ subscription_path=known_args.pubsub_subscription)
+
+ # Start feeder thread that reads URIs from GCS and fills Pub/Sub.
+ # Delay is used to allow the main streaming pipeline workers to start
+ # and autoscale before the feeder pipeline begins publishing messages.
+ threading.Thread(
+ target=lambda: (
+ time.sleep(known_args.feeder_start_delay_sec), run_load_pipeline(
+ known_args, pipeline_args)),
+ daemon=True).start()
+
+ # StandardOptions
+ pipeline_options = PipelineOptions(pipeline_args)
+ pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
+ pipeline_options.view_as(StandardOptions).streaming = (
+ known_args.mode == 'streaming')
+
+ # Build model handler with right-fitting batch size
+ desired_batch = pick_batch_size(known_args.inference_batch_size)
+
+ # Device
+ device = 'GPU' if known_args.device.upper() == 'GPU' else 'CPU'
+
+ tried = [64, 32, 16, 8] if desired_batch is None else [desired_batch]
+
+ model_handler = RightFittingPytorchModelHandlerTensor(
+ batch_sizes_to_try=tried,
+ image_size=known_args.image_size,
+ device=device,
+ model_class=lambda: create_timm_m(known_args.pretrained_model_name),
+ model_params={},
+ state_dict_path=known_args.model_state_dict_path,
+ inference_batch_size=tried[0],
+ )
+
+ pipeline = test_pipeline or beam.Pipeline(options=pipeline_options)
+
+ if known_args.mode == 'batch':
+ pcoll = (
+ pipeline
+ | 'ReadURIsBatch' >> beam.io.ReadFromText(known_args.input)
+ | 'FilterEmptyBatch' >> beam.Filter(lambda s: s.strip()))
+ else:
+ pcoll = (
+ pipeline
+ | 'ReadFromPubSub' >>
+ beam.io.ReadFromPubSub(subscription=known_args.pubsub_subscription)
+ | 'DecodeUTF8' >> beam.Map(lambda x: x.decode('utf-8'))
+ | 'Window' >> beam.WindowInto(
+ window.FixedWindows(known_args.window_sec),
+ trigger=beam.trigger.AfterProcessingTime(
+ known_args.trigger_proc_time_sec),
+ accumulation_mode=beam.trigger.AccumulationMode.DISCARDING,
+ allowed_lateness=0))
+
+ keyed = (
+ pcoll
+ | 'MakeKey' >> beam.ParDo(MakeKeyDoFn(input_mode=known_args.input_mode)))
+
+ preprocessed = (
+ keyed
+ | 'DecodePreprocess' >> beam.ParDo(
+ DecodePreprocessDoFn(
+ input_mode=known_args.input_mode,
+ image_size=known_args.image_size)))
+
+ to_infer = (
+ preprocessed
+ |
+ 'ToKeyedTensor' >> beam.Map(lambda kv: (kv[0], kv[1]["tensor"].float())))
+
+ run_inference = RunInference(KeyedModelHandler(model_handler))
+ if known_args.device.upper() == 'GPU':
+ run_inference = run_inference.with_resource_hints(
+ accelerator="type:nvidia-tesla-t4;count:1;install-nvidia-driver")
+ predictions = (
+ to_infer
+ | 'Reshuffle' >> beam.Reshuffle()
+ | 'RunInference' >> run_inference)
+
+ results = (
+ predictions
+ | 'PostProcess' >> beam.ParDo(
+ PostProcessDoFn(
+ top_k=known_args.top_k,
+ model_name=known_args.pretrained_model_name)))
+
+ method = (
+ beam.io.WriteToBigQuery.Method.FILE_LOADS if known_args.mode == 'batch'
+ else beam.io.WriteToBigQuery.Method.STREAMING_INSERTS)
+
+ if known_args.publish_to_big_query == 'true':
+ _ = (
+ results
+ | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
+ known_args.output_table,
+ schema=
+ 'image_id:STRING, model_name:STRING, topk:STRING, infer_ms:INT64',
+ write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
+ create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
+ method=method))
+
+ result = pipeline.run()
+ try:
+ result.wait_until_finish(duration=1800000) # 30 min
+ finally:
+ try:
+ result.cancel()
+ except Exception:
+ logging.debug("Failed to cancel pipeline result.", exc_info=True)
+
+ if known_args.mode == 'streaming':
+ cleanup_pubsub_resources(
+ project=known_args.project,
+ topic_path=known_args.pubsub_topic,
+ subscription_path=known_args.pubsub_subscription)
+
+ return result
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ run()
diff --git
a/sdks/python/apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
b/sdks/python/apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
new file mode 100644
index 00000000000..726cc147a0a
--- /dev/null
+++
b/sdks/python/apache_beam/ml/inference/pytorch_image_captioning_requirements.txt
@@ -0,0 +1,27 @@
+#
+# 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.
+#
+
+torch>=2.6.0,<2.8.0
+Pillow>=10.0.0
+numpy>=1.25.0
+transformers>=4.41.0,<5.0.0
+accelerate>=0.30.0
+tokenizers>=0.19.0
+safetensors>=0.4.3
+protobuf>=4.25.1
+requests>=2.31.0
+google-cloud-monitoring>=2.27.0
diff --git
a/sdks/python/apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
b/sdks/python/apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
new file mode 100644
index 00000000000..c3ce392d1a4
--- /dev/null
+++
b/sdks/python/apache_beam/ml/inference/pytorch_image_object_detection_requirements.txt
@@ -0,0 +1,24 @@
+#
+# 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.
+#
+
+torch>=2.2.0,<2.8.0
+torchvision>=0.17.0,<0.21.0
+Pillow>=10.0.0
+numpy>=1.25.0
+google-cloud-monitoring>=2.27.0
+protobuf>=4.25.1
+requests>=2.31.0
diff --git
a/sdks/python/apache_beam/ml/inference/pytorch_rightfit_requirements.txt
b/sdks/python/apache_beam/ml/inference/pytorch_rightfit_requirements.txt
new file mode 100644
index 00000000000..2b2916c577e
--- /dev/null
+++ b/sdks/python/apache_beam/ml/inference/pytorch_rightfit_requirements.txt
@@ -0,0 +1,26 @@
+#
+# 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.
+#
+
+torch>=2.2.0,<2.8.0
+torchvision>=0.17.0,<0.21.0
+timm>=1.0.7
+Pillow>=10.0.0
+numpy>=1.25.0
+google-cloud-pubsub>=2.15.0
+google-cloud-monitoring>=2.27.0
+protobuf>=4.25.1
+requests>=2.31.0
diff --git
a/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_image_captioning_benchmarks.py
b/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_image_captioning_benchmarks.py
new file mode 100644
index 00000000000..2960bfb7468
--- /dev/null
+++
b/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_image_captioning_benchmarks.py
@@ -0,0 +1,42 @@
+#
+# 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.
+#
+# pytype: skip-file
+
+import logging
+
+from apache_beam.examples.inference import pytorch_image_captioning
+from apache_beam.testing.load_tests.dataflow_cost_benchmark import
DataflowCostBenchmark
+
+
+class PytorchImageCaptioningBenchmarkTest(DataflowCostBenchmark):
+ def __init__(self):
+ self.metrics_namespace = 'BeamML_PyTorch'
+ super().__init__(
+ metrics_namespace=self.metrics_namespace,
+ pcollection='PostProcess.out0')
+
+ def test(self):
+ extra_opts = {}
+ extra_opts['input'] = self.pipeline.get_option('input_file')
+ self.result = pytorch_image_captioning.run(
+ self.pipeline.get_full_options_as_args(**extra_opts),
+ test_pipeline=self.pipeline)
+
+
+if __name__ == '__main__':
+ logging.basicConfig(level=logging.INFO)
+ PytorchImageCaptioningBenchmarkTest().run()
diff --git
a/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_image_object_detection_benchmarks.py
b/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_image_object_detection_benchmarks.py
new file mode 100644
index 00000000000..3bfd5daec18
--- /dev/null
+++
b/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_image_object_detection_benchmarks.py
@@ -0,0 +1,42 @@
+#
+# 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.
+#
+# pytype: skip-file
+
+import logging
+
+from apache_beam.examples.inference import pytorch_image_object_detection
+from apache_beam.testing.load_tests.dataflow_cost_benchmark import
DataflowCostBenchmark
+
+
+class PytorchImageObjectDetectionBenchmarkTest(DataflowCostBenchmark):
+ def __init__(self):
+ self.metrics_namespace = 'BeamML_PyTorch'
+ super().__init__(
+ metrics_namespace=self.metrics_namespace,
+ pcollection='PostProcess.out0')
+
+ def test(self):
+ extra_opts = {}
+ extra_opts['input'] = self.pipeline.get_option('input_file')
+ self.result = pytorch_image_object_detection.run(
+ self.pipeline.get_full_options_as_args(**extra_opts),
+ test_pipeline=self.pipeline)
+
+
+if __name__ == '__main__':
+ logging.basicConfig(level=logging.INFO)
+ PytorchImageObjectDetectionBenchmarkTest().run()
diff --git
a/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_imagenet_rightfit_benchmarks.py
b/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_imagenet_rightfit_benchmarks.py
new file mode 100644
index 00000000000..1528711a801
--- /dev/null
+++
b/sdks/python/apache_beam/testing/benchmarks/inference/pytorch_imagenet_rightfit_benchmarks.py
@@ -0,0 +1,42 @@
+#
+# 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.
+#
+# pytype: skip-file
+
+import logging
+
+from apache_beam.examples.inference import pytorch_imagenet_rightfit
+from apache_beam.testing.load_tests.dataflow_cost_benchmark import
DataflowCostBenchmark
+
+
+class PytorchImagenetRightfitBenchmarkTest(DataflowCostBenchmark):
+ def __init__(self):
+ self.metrics_namespace = 'BeamML_PyTorch'
+ super().__init__(
+ metrics_namespace=self.metrics_namespace,
+ pcollection='PostProcess.out0')
+
+ def test(self):
+ extra_opts = {}
+ extra_opts['input'] = self.pipeline.get_option('input_file')
+ self.result = pytorch_imagenet_rightfit.run(
+ self.pipeline.get_full_options_as_args(**extra_opts),
+ test_pipeline=self.pipeline)
+
+
+if __name__ == '__main__':
+ logging.basicConfig(level=logging.INFO)
+ PytorchImagenetRightfitBenchmarkTest().run()
diff --git a/website/www/site/content/en/performance/_index.md
b/website/www/site/content/en/performance/_index.md
index 347b7c1110c..db05afd4cbb 100644
--- a/website/www/site/content/en/performance/_index.md
+++ b/website/www/site/content/en/performance/_index.md
@@ -46,6 +46,14 @@ See the following pages for performance measures recorded
when running various B
## Streaming
- [PyTorch Sentiment Analysis Streaming DistilBERT
base](/performance/pytorchbertsentimentstreaming)
+- [PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
CPU](/performance/pytorchimagenetrightfitcpu)
+- [PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting
Exactly-once) CPU](/performance/pytorchimagenetrightfitoncecpu)
+- [PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
GPU](/performance/pytorchimagenetrightfitgpu)
+- [PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting
Exactly-once)](/performance/pytorchimagenetrightfitoncegpu)
+- [PyTorch Image Captioning BLIP + CLIP Streaming
CPU](/performance/pytorchimagecaptioningstreamingcpu)
+- [PyTorch Image Captioning BLIP + CLIP Streaming
GPU](/performance/pytorchimagecaptioningstreaminggpu)
+- [PyTorch Image Object Detection Faster R-CNN ResNet-50 Streaming
CPU](/performance/pytorchimageobjectdetectionstreamingcpu)
+- [PyTorch Image Object Detection Faster R-CNN ResNet-50 Streaming
GPU](/performance/pytorchimageobjectdetectionstreaminggpu)
- [Table Row Inference Sklearn
Streaming](/performance/tablerowinferencestreaming)
## Batch
@@ -59,6 +67,10 @@ See the following pages for performance measures recorded
when running various B
- [TensorFlow MNIST Image Classification](/performance/tensorflowmnist)
- [VLLM Gemma Batch Completion Tesla T4 GPU](/performance/vllmgemmabatchtesla)
- [Table Row Inference Sklearn Batch](/performance/tablerowinference)
+- [PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch
CPU](/performance/pytorchimageobjectdetectionbatchcpu)
+- [PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch
GPU](/performance/pytorchimageobjectdetectionbatchgpu)
+- [PyTorch Image Captioning BLIP + CLIP Batch
CPU](/performance/pytorchimagecaptioningbatchcpu)
+- [PyTorch Image Captioning BLIP + CLIP Batch
GPU](/performance/pytorchimagecaptioningbatchgpu)
- [MLTransform Generate Vocab (batch)](/performance/mltransformvocab)
- [MLTransform One-Hot Encoding](/performance/mltransformonehot)
- [MLTransform Text Embedding](/performance/mltransform-text-embedding)
diff --git
a/website/www/site/content/en/performance/pytorchimagecaptioningbatchcpu/_index.md
b/website/www/site/content/en/performance/pytorchimagecaptioningbatchcpu/_index.md
new file mode 100644
index 00000000000..d5d5f094e55
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagecaptioningbatchcpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Captioning BLIP + CLIP Batch CPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Captioning BLIP + CLIP Batch CPU
+
+**Model**: PyTorch Image Captioning — BLIP (candidate generation) + CLIP
(ranking)
+**Accelerator**: CPU only
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This batch pipeline performs image captioning using a multi-model open-source
PyTorch approach.
+It first generates multiple candidate captions per image using a BLIP model,
then ranks these candidates with a CLIP model based on image-text similarity.
+
+The following graphs show various metrics when running PyTorch Image
Captioning BLIP + CLIP Batch CPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningbatchcpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagecaptioningbatchcpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningbatchcpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagecaptioningbatchgpu/_index.md
b/website/www/site/content/en/performance/pytorchimagecaptioningbatchgpu/_index.md
new file mode 100644
index 00000000000..d0d12bf3f54
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagecaptioningbatchgpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Captioning BLIP + CLIP Batch GPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Captioning BLIP + CLIP Batch GPU
+
+**Model**: PyTorch Image Captioning — BLIP (candidate generation) + CLIP
(ranking)
+**Accelerator**: Tesla T4 GPU
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This batch pipeline performs image captioning using a multi-model open-source
PyTorch approach.
+It first generates multiple candidate captions per image using a BLIP model,
then ranks these candidates with a CLIP model based on image-text similarity.
+
+The following graphs show various metrics when running PyTorch Image
Captioning BLIP + CLIP Batch GPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningbatchgpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagecaptioningbatchgpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningbatchgpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagecaptioningstreamingcpu/_index.md
b/website/www/site/content/en/performance/pytorchimagecaptioningstreamingcpu/_index.md
new file mode 100644
index 00000000000..41223f1bf0b
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagecaptioningstreamingcpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Captioning BLIP + CLIP Streaming CPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Captioning BLIP + CLIP Streaming CPU
+
+**Model**: PyTorch Image Captioning — BLIP (candidate generation) + CLIP
(ranking)
+**Accelerator**: CPU only
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs image captioning using a multi-model
open-source PyTorch approach.
+It first generates multiple candidate captions per image using a BLIP model,
then ranks these candidates with a CLIP model based on image-text similarity.
+
+The following graphs show various metrics when running PyTorch Image
Captioning BLIP + CLIP Streaming CPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningstreamingcpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagecaptioningstreamingcpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningstreamingcpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagecaptioningstreaminggpu/_index.md
b/website/www/site/content/en/performance/pytorchimagecaptioningstreaminggpu/_index.md
new file mode 100644
index 00000000000..b7694baa82e
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagecaptioningstreaminggpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Captioning BLIP + CLIP Streaming GPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Captioning BLIP + CLIP Streaming GPU
+
+**Model**: PyTorch Image Captioning — BLIP (candidate generation) + CLIP
(ranking)
+**Accelerator**: Tesla T4 GPU
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs image captioning using a multi-model
open-source PyTorch approach.
+It first generates multiple candidate captions per image using a BLIP model,
then ranks these candidates with a CLIP model based on image-text similarity.
+
+The following graphs show various metrics when running PyTorch Image
Captioning BLIP + CLIP Streaming GPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_captioning.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningstreaminggpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagecaptioningstreaminggpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagecaptioningstreaminggpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagenetrightfitcpu/_index.md
b/website/www/site/content/en/performance/pytorchimagenetrightfitcpu/_index.md
new file mode 100644
index 00000000000..9b7b26d9b3e
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagenetrightfitcpu/_index.md
@@ -0,0 +1,43 @@
+---
+title: "PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
CPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
+
+**Model**: PyTorch Image Classification — EfficientNet-B0 (pretrained on
ImageNet)
+**Accelerator**: CPU only
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs image classification using an open-source
PyTorch EfficientNet-B0 model.
+It reads image URIs from Pub/Sub, decodes and preprocesses them in parallel,
and runs inference with adaptive batch sizing.
+
+The following graphs show various metrics when running the PyTorch Image
Classification EfficientNet-B0 Streaming (Right-fitting) CPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitcpu" read_or_write="write"
section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagenetrightfitcpu" read_or_write="write"
section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitcpu" read_or_write="write"
section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagenetrightfitgpu/_index.md
b/website/www/site/content/en/performance/pytorchimagenetrightfitgpu/_index.md
new file mode 100644
index 00000000000..fefd149b841
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagenetrightfitgpu/_index.md
@@ -0,0 +1,43 @@
+---
+title: "PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
GPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
+
+**Model**: PyTorch Image Classification — EfficientNet-B0 (pretrained on
ImageNet)
+**Accelerator**: Tesla T4 GPU (right-fitted batch size 64 → 32 → 16 → 8)
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs image classification using an open-source
PyTorch EfficientNet-B0 model optimized for T4 GPUs.
+It reads image URIs from Pub/Sub, decodes and preprocesses them in parallel,
and runs inference with adaptive batch sizing for optimal GPU utilization.
+
+The following graphs show various metrics when running the PyTorch Image
Classification EfficientNet-B0 Streaming (Right-fitting) GPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitgpu" read_or_write="write"
section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagenetrightfitgpu" read_or_write="write"
section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitgpu" read_or_write="write"
section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagenetrightfitoncecpu/_index.md
b/website/www/site/content/en/performance/pytorchimagenetrightfitoncecpu/_index.md
new file mode 100644
index 00000000000..8e0cf82c349
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagenetrightfitoncecpu/_index.md
@@ -0,0 +1,44 @@
+---
+title: "PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting,
Exactly-once) CPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
+
+**Model**: PyTorch Image Classification — EfficientNet-B0 (pretrained on
ImageNet)
+**Accelerator**: CPU only
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs image classification using an open-source
PyTorch EfficientNet-B0 model.
+It reads image URIs from Pub/Sub, decodes and preprocesses them in parallel,
and runs inference with adaptive batch sizing.
+The pipeline ensures exactly-once semantics via stateful deduplication and
idempotent BigQuery writes, allowing stable and reproducible performance
measurements under continuous load.
+
+The following graphs show various metrics when running the PyTorch Image
Classification EfficientNet-B0 Streaming (Right-fitting, Exactly-once) CPU
pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitoncecpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagenetrightfitoncecpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitoncecpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimagenetrightfitoncegpu/_index.md
b/website/www/site/content/en/performance/pytorchimagenetrightfitoncegpu/_index.md
new file mode 100644
index 00000000000..85a55b27955
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimagenetrightfitoncegpu/_index.md
@@ -0,0 +1,44 @@
+---
+title: "PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting,
Exactly-once) GPU Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Classification EfficientNet-B0 Streaming (Right-fitting)
+
+**Model**: PyTorch Image Classification — EfficientNet-B0 (pretrained on
ImageNet)
+**Accelerator**: Tesla T4 GPU (right-fitted batch size 64 → 32 → 16 → 8)
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs image classification using an open-source
PyTorch EfficientNet-B0 model optimized for T4 GPUs.
+It reads image URIs from Pub/Sub, decodes and preprocesses them in parallel,
and runs inference with adaptive batch sizing for optimal GPU utilization.
+The pipeline ensures exactly-once semantics via stateful deduplication and
idempotent BigQuery writes, allowing stable and reproducible performance
measurements under continuous load.
+
+The following graphs show various metrics when running the PyTorch Image
Classification EfficientNet-B0 Streaming (Right-fitting, Exactly-once) GPU
pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_imagenet_rightfit.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitoncegpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimagenetrightfitoncegpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimagenetrightfitoncegpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimageobjectdetectionbatchcpu/_index.md
b/website/www/site/content/en/performance/pytorchimageobjectdetectionbatchcpu/_index.md
new file mode 100644
index 00000000000..a2ef39a2cf3
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimageobjectdetectionbatchcpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch CPU
Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch CPU
+
+**Model**: PyTorch Image Object Detection — Faster R-CNN ResNet-50 FPN
(pretrained on COCO)
+**Accelerator**: CPU only
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This batch pipeline performs object detection using an open-source PyTorch
Faster R-CNN ResNet-50 FPN model on CPU.
+It reads image URIs from GCS, decodes and preprocesses images, and runs
batched inference with a fixed batch size.
+
+The following graphs show various metrics when running PyTorch Image Object
Detection Faster R-CNN ResNet-50 Batch CPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionbatchcpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimageobjectdetectionbatchcpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionbatchcpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimageobjectdetectionbatchgpu/_index.md
b/website/www/site/content/en/performance/pytorchimageobjectdetectionbatchgpu/_index.md
new file mode 100644
index 00000000000..f6d6107d19f
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimageobjectdetectionbatchgpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch GPU
Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Object Detection Faster R-CNN ResNet-50 Batch GPU
+
+**Model**: PyTorch Image Object Detection — Faster R-CNN ResNet-50 FPN
(pretrained on COCO)
+**Accelerator**: Tesla T4 GPU (fixed batch size)
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This batch pipeline performs object detection using an open-source PyTorch
Faster R-CNN ResNet-50 FPN model on GPU.
+It reads image URIs from GCS, decodes and preprocesses images, and runs
batched inference with a fixed batch size to measure stable GPU performance.
+
+The following graphs show various metrics when running PyTorch Image Object
Detection Faster R-CNN ResNet-50 Batch GPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionbatchgpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimageobjectdetectionbatchgpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionbatchgpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimageobjectdetectionstreamingcpu/_index.md
b/website/www/site/content/en/performance/pytorchimageobjectdetectionstreamingcpu/_index.md
new file mode 100644
index 00000000000..b866d2893cd
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimageobjectdetectionstreamingcpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Object Detection Faster R-CNN ResNet-50 Streaming CPU
Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Object Detection Faster R-CNN ResNet-50 Streaming CPU
+
+**Model**: PyTorch Image Object Detection — Faster R-CNN ResNet-50 FPN
(pretrained on COCO)
+**Accelerator**: CPU only
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs object detection using an open-source PyTorch
Faster R-CNN ResNet-50 FPN model on CPU.
+It reads image URIs from GCS, decodes and preprocesses images, and runs
inference.
+
+The following graphs show various metrics when running PyTorch Image Object
Detection Faster R-CNN ResNet-50 Streaming CPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionstreamingcpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimageobjectdetectionstreamingcpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionstreamingcpu"
read_or_write="write" section="date" >}}
diff --git
a/website/www/site/content/en/performance/pytorchimageobjectdetectionstreaminggpu/_index.md
b/website/www/site/content/en/performance/pytorchimageobjectdetectionstreaminggpu/_index.md
new file mode 100644
index 00000000000..443f5e99e76
--- /dev/null
+++
b/website/www/site/content/en/performance/pytorchimageobjectdetectionstreaminggpu/_index.md
@@ -0,0 +1,41 @@
+---
+title: "PyTorch Image Object Detection Faster R-CNN ResNet-50 Streaming GPU
Performance"
+---
+
+<!--
+Licensed 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.
+-->
+
+# PyTorch Image Object Detection Faster R-CNN ResNet-50 Streaming GPU
+
+**Model**: PyTorch Image Object Detection — Faster R-CNN ResNet-50 FPN
(pretrained on COCO)
+**Accelerator**: Tesla T4 GPU
+**Host**: 50 × n1-standard-4 (4 vCPUs, 15 GB RAM)
+
+This streaming pipeline performs object detection using an open-source PyTorch
Faster R-CNN ResNet-50 FPN model on GPU.
+It reads image URIs from GCS, decodes and preprocesses images, and runs
inference.
+
+The following graphs show various metrics when running PyTorch Image Object
Detection Faster R-CNN ResNet-50 Streaming GPU pipeline.
+See the [glossary](/performance/glossary) for definitions.
+
+Full pipeline implementation is available
[here](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_image_object_detection.py).
+
+## What is the estimated cost to run the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionstreaminggpu"
read_or_write="write" section="cost" >}}
+
+## How has various metrics changed when running the pipeline for different
Beam SDK versions?
+
+{{< performance_looks io="pytorchimageobjectdetectionstreaminggpu"
read_or_write="write" section="version" >}}
+
+## How has various metrics changed over time when running the pipeline?
+
+{{< performance_looks io="pytorchimageobjectdetectionstreaminggpu"
read_or_write="write" section="date" >}}
diff --git a/website/www/site/data/performance.yaml
b/website/www/site/data/performance.yaml
index c27dcc22f23..f0292fab747 100644
--- a/website/www/site/data/performance.yaml
+++ b/website/www/site/data/performance.yaml
@@ -283,21 +283,197 @@ looks:
title: AvgThroughputBytesPerSec by Version
- id: P7wKZy6tQFWbbDfm4HzfCJnsQrVgfGsJ
title: AvgThroughputElementsPerSec by Version
- mltransformvocab:
+ pytorchimagenetrightfitcpu:
write:
- folder: 107
+ folder: 92
cost:
- - id: CccNCHQ23Js2tmxTDYVnjxwfx2zSFSjY
+ - id: zJhxrMmxJ3zVHH5WZnDQqcBHdFDrBhxK
title: RunTime and EstimatedCost
date:
- - id: n7Gn7c7KSW52ZgXCQXndHgWtjVPK8W33
+ - id: RybzxZdkXJg6PzQZkBcfxTJkByTf3ZV5
title: AvgThroughputBytesPerSec by Date
- - id: 3tcg6cSh6tBg6FmpTxgzyMtvFqWKkQn3
+ - id: xTVYPytQVH7zXYz7SvphnRV4nQcxCddp
title: AvgThroughputElementsPerSec by Date
version:
- - id: tPYwJngPDBsKjK3DNgGHGsCyTpxdfmTB
+ - id: dGN6Zr6rh7DfnRtTDCN6GHcNfhSkrbCq
title: AvgThroughputBytesPerSec by Version
- - id: cC75NnCbQT3mQmKVHtDxzptXpwPb64qz
+ - id: VJMWrZh3jXk2mqCZk4NQn3tBrHgGWqnC
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagenetrightfitoncecpu:
+ write:
+ folder: 97
+ cost:
+ - id: DhsvrRxpby2F25jbj8NPVR6FhzKgsHBP
+ title: RunTime and EstimatedCost
+ date:
+ - id: sntwpmZjMVT57kvG8KM8rzDjBKsrRTF7
+ title: AvgThroughputBytesPerSec by Date
+ - id: 5GJHCVH65y2zmnrZ3QqmYqk8vKfPD4D6
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: rz24FQ24rz3HtFmpzK2rKD4zFT5kpG46
+ title: AvgThroughputBytesPerSec by Version
+ - id: jZkcSdHbvmGXy8f37RShNvgGv6HVN7Dy
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagenetrightfitgpu:
+ write:
+ folder: 98
+ cost:
+ - id: sMgzq4N8kJfjSjhSzVDzmXSHbXKwCy4r
+ title: RunTime and EstimatedCost
+ date:
+ - id: t3D6hqq7BywmvjBmxtWCvRMsFFMgPBg7
+ title: AvgThroughputBytesPerSec by Date
+ - id: MRptKCFsnTKcxPgVKKVvs7c8XVTf6MSN
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: v6jHqBc8WvCPspgdxQdjSZ6bCgqqPvZ3
+ title: AvgThroughputBytesPerSec by Version
+ - id: DhGNQDmRDhYSCDgXDsx6BtCcDfwKCWhb
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagenetrightfitoncegpu:
+ write:
+ folder: 99
+ cost:
+ - id: YbwBRmMVBCxPzjm22vWvJ2fDG5mmM3BR
+ title: RunTime and EstimatedCost
+ date:
+ - id: WtCZn2CrRrDdygbxQhfVhNM8F7NRRpsW
+ title: AvgThroughputBytesPerSec by Date
+ - id: RBMNkqksbcN3WcwCxz4B5mxVq5Y8yk2x
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: 48tzWvydG73Y3QwPJwQzYCkzdDnpF2T7
+ title: AvgThroughputBytesPerSec by Version
+ - id: 3VDCrNY6q8FyS28YRrxjJtMBYkxKRdBw
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimageobjectdetectionbatchgpu:
+ write:
+ folder: 93
+ cost:
+ - id: XGPVSYhVbZGJHQCtMPW4nRGynxXpdzdh
+ title: RunTime and EstimatedCost
+ date:
+ - id: ZW48KBPBxShGgWx53vjfvgqp6cVmHCNB
+ title: AvgThroughputBytesPerSec by Date
+ - id: BjMWg26F3gNHQyZSsg8HjhTg3mCh6jFJ
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: kbF7Gnqjsjnvh3MKvYszSgWFbgDYYTWR
+ title: AvgThroughputBytesPerSec by Version
+ - id: cTxrpY3KGrCb35dq7fnjvdsDnd7t85pJ
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimageobjectdetectionbatchcpu:
+ write:
+ folder: 100
+ cost:
+ - id: QbMVTdnxzmZzfymSrKsGTvyqFTMpwqy7
+ title: RunTime and EstimatedCost
+ date:
+ - id: v6H8sqn7X66j5g6qpgC6B57z624cCfDM
+ title: AvgThroughputBytesPerSec by Date
+ - id: KBMSpWzB5RgdYsTdZxGCXDYbWpskVqGS
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: GWz5VDrMTxSCrFqQwk2MyPHMwBHC253c
+ title: AvgThroughputBytesPerSec by Version
+ - id: NBWHNG8GmGHp8Bmg25wqg3tP8jWdP5PW
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimageobjectdetectionstreaminggpu:
+ write:
+ folder: 101
+ cost:
+ - id: SfgfDsbRVk67XMnKZShq3xVBFghN7GSv
+ title: RunTime and EstimatedCost
+ date:
+ - id: MtMJZTTrkgXZz24zbQbpzwYNjWwcfhqw
+ title: AvgThroughputBytesPerSec by Date
+ - id: kKncXDm5MKvfwVzFyXqQ4TpfR5VdT5z7
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: ZWQ7vnsXYb5PDr5yDFjM3QP8gkkmdgdQ
+ title: AvgThroughputBytesPerSec by Version
+ - id: cTmDCbdCQyrsK3fS333vmR3BwCwrGZgp
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimageobjectdetectionstreamingcpu:
+ write:
+ folder: 102
+ cost:
+ - id: FHgY6JfxRc23jJ6THkb7sPVw2HSw4kgs
+ title: RunTime and EstimatedCost
+ date:
+ - id: G5DNH6vXpGVnYP8ZGMwppdGvgvndM8kn
+ title: AvgThroughputBytesPerSec by Date
+ - id: CGR43NrksWDscGNWCKsVGCwc3RxmvGhD
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: wGGVRGxBjXxQf3yvynSFtcDCVCqKzByy
+ title: AvgThroughputBytesPerSec by Version
+ - id: M5nWYF76fWPnbTH6QkzWqzRV8fSWxtWD
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagecaptioningbatchgpu:
+ write:
+ folder: 94
+ cost:
+ - id: nhYggnCYJzfgDyTXZRGJwpSyYKvBpfhV
+ title: RunTime and EstimatedCost
+ date:
+ - id: qzjkYQJCrxjc8GPcBF7dFJSFdTXQKxmR
+ title: AvgThroughputBytesPerSec by Date
+ - id: kgkdbBCbkhDHpb2rHNGQDxcwGqzykwqD
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: 4p7b4HWVMRC8HXDZYMXbVYf2VPzYQ2kz
+ title: AvgThroughputBytesPerSec by Version
+ - id: xQZRDYJRvmV7qMhb3j2XTrPmC8TP4DTr
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagecaptioningbatchcpu:
+ write:
+ folder: 103
+ cost:
+ - id: M4vTdbrK6BxJXdXcnR7WPwW7VF4Fmhbb
+ title: RunTime and EstimatedCost
+ date:
+ - id: T23Fty8X6dF6DFhxSy32ymSwqz64fPXT
+ title: AvgThroughputBytesPerSec by Date
+ - id: Wm2qdmKgHbzfJbW7vNVfrGtQWV4xGCx3
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: wnQqxHgwJstnV4fPCkNpvMxJpyPG4326
+ title: AvgThroughputBytesPerSec by Version
+ - id: kPMV7d8Htpp6CT8q5hC6NF5DbTRwBQcJ
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagecaptioningstreaminggpu:
+ write:
+ folder: 104
+ cost:
+ - id: rgrPw4ZdZjCD8cvKqdvdPGGtPJgWfNK5
+ title: RunTime and EstimatedCost
+ date:
+ - id: jwpm4S36mGXPrY2nfpWrCvDHN6Nsrnpq
+ title: AvgThroughputBytesPerSec by Date
+ - id: QmY5QVTpndNTBJdqWjMxxD4zQkZqn6Xs
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: htD3YwWcTPYcRk4wqtqHshnt7hrgyrfk
+ title: AvgThroughputBytesPerSec by Version
+ - id: rv3QbsbxBV68MWG8hdjhnH84zJsPgJQh
+ title: AvgThroughputElementsPerSec by Version
+ pytorchimagecaptioningstreamingcpu:
+ write:
+ folder: 105
+ cost:
+ - id: YkFKPzV2Rvgy5hmmx2KPyJCHhbxhSjzj
+ title: RunTime and EstimatedCost
+ date:
+ - id: 5WStCthVw8gtCdKrY5mhQf8T6N45GGqH
+ title: AvgThroughputBytesPerSec by Date
+ - id: ZnJj4QFrd3dwmsmv3xdNKytPfbW7j88C
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: VHqjNCVNScFqHF8d65khMDGxfSPrBrMh
+ title: AvgThroughputBytesPerSec by Version
+ - id: ykZJhrhHDzyPBSTxyNcdYG7yw5qs3dYj
title: AvgThroughputElementsPerSec by Version
mltransformonehot:
write:
@@ -315,6 +491,22 @@ looks:
title: AvgThroughputBytesPerSec by Version
- id: Cn5FsXkdy2ZXCxCJshSCxcjsTW3TXf3c
title: AvgThroughputElementsPerSec by Version
+ mltransformvocab:
+ write:
+ folder: 107
+ cost:
+ - id: CccNCHQ23Js2tmxTDYVnjxwfx2zSFSjY
+ title: RunTime and EstimatedCost
+ date:
+ - id: n7Gn7c7KSW52ZgXCQXndHgWtjVPK8W33
+ title: AvgThroughputBytesPerSec by Date
+ - id: 3tcg6cSh6tBg6FmpTxgzyMtvFqWKkQn3
+ title: AvgThroughputElementsPerSec by Date
+ version:
+ - id: tPYwJngPDBsKjK3DNgGHGsCyTpxdfmTB
+ title: AvgThroughputBytesPerSec by Version
+ - id: cC75NnCbQT3mQmKVHtDxzptXpwPb64qz
+ title: AvgThroughputElementsPerSec by Version
mltransform-text-embedding:
write:
folder: 109
