[Bug]: YAML Transform Provider with AnomalyDetection

[charlespnh]

What happened?

I have a custom anomaly detection model trained using the PyOD library, and I want to use it as part of my YAML pipeline. Beam's anomaly detection module offers a way to use a custom offline model (see this example notebook )

I implemented my own transform following that example, and exposed it through Provider:

import apache_beam as beam
from apache_beam.ml.anomaly.detectors.pyod_adapter import PyODFactory
from apache_beam.ml.anomaly.transforms import AnomalyDetection


class KNN(beam.PTransform):
  def __init__(self, model_artifact_path):
    self.model_artifact_path = model_artifact_path
    self.model = PyODFactory.create_detector(
        self.model_artifact_path,
        model_id="knn",
    )

  def expand(self, pcoll):
    return (
        pcoll
        | beam.Map(lambda x: x.embedding)
        | AnomalyDetection(detector=self.model)
        | beam.Map(lambda x: beam.Row(
            example=x.example,
            predictions=[pred.__dict__ for pred in x.predictions]))
    )

The following YAML pipeline:

pipeline:
  type: chain
  transforms:
    - type: ReadFromBigQuery
      name: ReadFromBigQuery
      config:
        table: "apache-beam-testing.charlesnguyen.test"
        fields: [embedding]

    - type: KNN
      name: KNN
      config:
        model_artifact_path: "gs://apache-beam-testing-charlesng/batch-log-analysis/knn_model.pkl"
#        model_artifact_path: ./knn_model.pkl

    - type: LogForTesting


providers:
  - type: pythonPackage
    config:
      packages:
        - ./dist/transform_provider-0.1.0.tar.gz
    transforms:
      KNN: "transform_provider.KNN"

options:
  yaml_experimental_features: [ 'ML' ]

... gives the error:

  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
    return transform.expand(input)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/transforms/external.py", line 824, in expand
    response = service.Expand(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/.virtualenvs/env/lib/python3.11/site-packages/grpc/_channel.py", line 1181, in __call__
    return _end_unary_response_blocking(state, call, False, None)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/.virtualenvs/env/lib/python3.11/site-packages/grpc/_channel.py", line 1006, in _end_unary_response_blocking
    raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNAVAILABLE
        details = "Socket closed"                                                                                                                                             debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"Socket closed", grpc_status:14, created_time:"2025-08-05T14:59:02.47977-04:00"}"
>

But if I have the knn_model.pkl model locally and specify model_artifact_path argument with the local path to the model instead of gcs path, then it works.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[charlespnh]

CC @chamikaramj

Also CC @shunping for this AnomalyDetection use case. I don't know if there is a better way to use AnomalyDetection with custom offline model in a YAML pipeline than this.

[shunping]

Very weird. The error is actually raised when trying to expand the external transform (ReadFromBigQuery) but not Anomaly Detection.

[chamikaramj]

I think this error simply means that you are trying to use the transform through a Python expansion service, but that service is not up.

[chamikaramj]

Started at the location below.

beam/sdks/python/apache_beam/yaml/yaml_provider.py

Line 439 in 720f961

PypiExpansionService([

Is there another error that indicate Python expansion service failing to start ?

[charlespnh]

Started at the location below.

beam/sdks/python/apache_beam/yaml/yaml_provider.py

Line 439 in 720f961

PypiExpansionService([
Is there another error that indicate Python expansion service failing to start ?

Full log message:

INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
Building pipeline...
WARNING:root:Could not load ML transform module apache_beam.ml.transforms.embeddings.open_ai: No module named 'openai'.  Please install the necessary module dependencies
WARNING:root:Could not load ML transform module apache_beam.ml.transforms.embeddings.tensorflow_hub: No module named 'tensorflow'.  Please install the necessary module dependencies
INFO:apache_beam.yaml.yaml_transform:Expanding "ReadFromBigQuery" at line 23
WARNING:apache_beam.transforms.core:('No iterator is returned by the process method in %s.', <class 'apache_beam.io.gcp.bigquery_read_internal._PassThroughThenCleanupTempDatasets.expand.<locals>.CleanUpProjects'>)
INFO:apache_beam.yaml.yaml_transform:Expanding "KNN" at line 29
INFO:apache_beam.utils.subprocess_server:Starting service with ['/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/bin/python' '-m' 'apache_beam.runners.portability.expansion_service_main' '--port' '62522' '--fully_qualified_name_glob=*' '--pickle_library=cloudpickle' '--requirements_file=/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964-requirements.txt' '--serve_loopback_worker']
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.portability.stager:Executing command: ['/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/bin/python', '-m', 'pip', 'download', '--dest', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/dataflow-requirements-cache', '-r', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/tmp_kdid1w4/tmp_requirements.txt', '--exists-action', 'i', '--no-deps', '--implementation', 'cp', '--abi', 'cp311', '--platform', 'manylinux2014_x86_64']
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.portability.stager:Executing command: ['/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/bin/python', '-m', 'pip', 'download', '--dest', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/dataflow-requirements-cache', '-r', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/tmpc8_q_frj/tmp_requirements.txt', '--exists-action', 'i', '--no-deps', '--implementation', 'cp', '--abi', 'cp311', '--platform', 'manylinux2014_x86_64']
INFO:apache_beam.utils.subprocess_server:INFO:__main__:Listening for expansion requests at 62522
INFO:apache_beam.utils.subprocess_server:ERROR:grpc._server:Exception calling application: Method not implemented!
INFO:apache_beam.utils.subprocess_server:Traceback (most recent call last):
INFO:apache_beam.utils.subprocess_server:  File "/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/lib/python3.11/site-packages/grpc/_server.py", line 610, in _call_behavior
INFO:apache_beam.utils.subprocess_server:    response_or_iterator = behavior(argument, context)
INFO:apache_beam.utils.subprocess_server:                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
INFO:apache_beam.utils.subprocess_server:  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/portability/api/org/apache/beam/model/job_management/v1/beam_expansion_api_pb2_grpc.py", line 46, in DiscoverSchemaTransform
INFO:apache_beam.utils.subprocess_server:    raise NotImplementedError('Method not implemented!')
INFO:apache_beam.utils.subprocess_server:NotImplementedError: Method not implemented!
INFO:apache_beam.utils.subprocess_server:INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
INFO:apache_beam.utils.subprocess_server:WARNING: All log messages before absl::InitializeLog() is called are written to STDERR
INFO:apache_beam.utils.subprocess_server:I0000 00:00:1754420342.099952 8355449 fork_posix.cc:77] Other threads are currently calling into gRPC, skipping fork() handlers
Traceback (most recent call last):
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 506, in expand_leaf_transform
    outputs = inputs | scope.unique_name(spec, ptransform) >> ptransform
              ~~~~~~~^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/pvalue.py", line 138, in __or__
    return self.pipeline.apply(ptransform, self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/pipeline.py", line 722, in apply
    return self.apply(
           ^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/pipeline.py", line 733, in apply
    return self.apply(transform, pvalueish)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/pipeline.py", line 813, in apply
    pvalueish_result = self.runner.apply(transform, pvalueish, self._options)
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/runners/runner.py", line 191, in apply
    return self.apply_PTransform(transform, input, options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/runners/runner.py", line 195, in apply_PTransform
    return transform.expand(input)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/transforms/external.py", line 824, in expand
    response = service.Expand(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/.virtualenvs/env/lib/python3.11/site-packages/grpc/_channel.py", line 1181, in __call__
    return _end_unary_response_blocking(state, call, False, None)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/.virtualenvs/env/lib/python3.11/site-packages/grpc/_channel.py", line 1006, in _end_unary_response_blocking
    raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNAVAILABLE
        details = "Socket closed"                                                                                                                                             debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"Socket closed", grpc_status:14, created_time:"2025-08-05T14:59:02.47977-04:00"}"
>

The above exception was the direct cause of the following exception:
                                                                                                                                                                      Traceback (most recent call last):
  File "<frozen runpy>", line 198, in _run_module_as_main
  File "<frozen runpy>", line 88, in _run_code
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/main.py", line 297, in <module>
    run()
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/main.py", line 148, in run
    constructor(p)
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/main.py", line 281, in constructor
    yaml_transform.expand_pipeline(
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 1137, in expand_pipeline
    providers or {})).expand(root)
                      ^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 1098, in expand
    result = expand_transform(
             ^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 478, in expand_transform
    return expand_composite_transform(spec, scope)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 561, in expand_composite_transform
    return transform.expand(None)
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 544, in expand
    inner_scope.compute_all()
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 205, in compute_all
    self.compute_outputs(transform_id)
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 103, in wrapper
    self._cache[key] = func(self, *args)
                       ^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 241, in compute_outputs
    return expand_transform(self._transforms_by_uuid[transform_id], self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 480, in expand_transform
    return expand_leaf_transform(spec, scope)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/yaml/yaml_transform.py", line 508, in expand_leaf_transform
    raise ValueError(
ValueError: Error applying transform "KNN" at line 29: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNAVAILABLE
        details = "Socket closed"
        debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"Socket closed", grpc_status:14, created_time:"2025-08-05T14:59:02.47977-04:00"}"
>

[chamikaramj]

It's the one below.

INFO:apache_beam.utils.subprocess_server:ERROR:grpc._server:Exception calling application: Method not implemented!
INFO:apache_beam.utils.subprocess_server:Traceback (most recent call last):
INFO:apache_beam.utils.subprocess_server:  File "/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/lib/python3.11/site-packages/grpc/_server.py", line 610, in _call_behavior
INFO:apache_beam.utils.subprocess_server:    response_or_iterator = behavior(argument, context)
INFO:apache_beam.utils.subprocess_server:                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
INFO:apache_beam.utils.subprocess_server:  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/portability/api/org/apache/beam/model/job_management/v1/beam_expansion_api_pb2_grpc.py", line 46, in DiscoverSchemaTransform
INFO:apache_beam.utils.subprocess_server:    raise NotImplementedError('Method not implemented!')
INFO:apache_beam.utils.subprocess_server:NotImplementedError: Method not implemented!

[charlespnh]

The weird thing is, if I point model_artifact_path to the local path of the model ./knn_model.pkl instead of gcs path, then it works... This error INFO:apache_beam.utils.subprocess_server:NotImplementedError: Method not implemented! occurs in both cases, though.

Here is what the log looks like when I use the local path to the model instead:

INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
Building pipeline...
WARNING:root:Could not load ML transform module apache_beam.ml.transforms.embeddings.open_ai: No module named 'openai'.  Please install the necessary module dependencies
WARNING:root:Could not load ML transform module apache_beam.ml.transforms.embeddings.tensorflow_hub: No module named 'tensorflow'.  Please install the necessary module dependencies
INFO:apache_beam.yaml.yaml_transform:Expanding "ReadFromBigQuery" at line 23
WARNING:apache_beam.transforms.core:('No iterator is returned by the process method in %s.', <class 'apache_beam.io.gcp.bigquery_read_internal._PassThroughThenCleanupTempDatasets.expand.<locals>.CleanUpProjects'>)
INFO:apache_beam.yaml.yaml_transform:Expanding "KNN" at line 31
INFO:apache_beam.utils.subprocess_server:Starting service with ['/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/bin/python' '-m' 'apache_beam.runners.portability.expansion_service_main' '--port' '62974' '--fully_qualified_name_glob=*' '--pickle_library=cloudpickle' '--requirements_file=/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964-requirements.txt' '--serve_loopback_worker']
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.portability.stager:Executing command: ['/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/bin/python', '-m', 'pip', 'download', '--dest', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/dataflow-requirements-cache', '-r', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/tmpp_2wzrpt/tmp_requirements.txt', '--exists-action', 'i', '--no-deps', '--implementation', 'cp', '--abi', 'cp311', '--platform', 'manylinux2014_x86_64']
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.portability.stager:Executing command: ['/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/bin/python', '-m', 'pip', 'download', '--dest', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/dataflow-requirements-cache', '-r', '/var/folders/c9/3ylgx85x553b8451_158qxpr0000gn/T/tmpa4ugtyer/tmp_requirements.txt', '--exists-action', 'i', '--no-deps', '--implementation', 'cp', '--abi', 'cp311', '--platform', 'manylinux2014_x86_64']
INFO:apache_beam.utils.subprocess_server:INFO:__main__:Listening for expansion requests at 62974
INFO:apache_beam.utils.subprocess_server:ERROR:grpc._server:Exception calling application: Method not implemented!
INFO:apache_beam.utils.subprocess_server:Traceback (most recent call last):
INFO:apache_beam.utils.subprocess_server:  File "/Users/charlesnguyen/.apache_beam/cache/venvs/5316be40511ca369dcddc7cc8c711c62d053c3cdfda7fc4168fbb4232ba67964/lib/python3.11/site-packages/grpc/_server.py", line 610, in _call_behavior
INFO:apache_beam.utils.subprocess_server:    response_or_iterator = behavior(argument, context)
INFO:apache_beam.utils.subprocess_server:                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
INFO:apache_beam.utils.subprocess_server:  File "/Users/charlesnguyen/Code/beam/sdks/python/apache_beam/portability/api/org/apache/beam/model/job_management/v1/beam_expansion_api_pb2_grpc.py", line 46, in DiscoverSchemaTransform
INFO:apache_beam.utils.subprocess_server:    raise NotImplementedError('Method not implemented!')
INFO:apache_beam.utils.subprocess_server:NotImplementedError: Method not implemented!
INFO:apache_beam.utils.subprocess_server:INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
INFO:apache_beam.yaml.yaml_transform:Expanding "LogForTesting" at line 37
Running pipeline...
INFO:apache_beam.runners.worker.statecache:Creating state cache with size 104857600
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting control server on port 62992
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting data server on port 62993
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting state server on port 62994
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting logging server on port 62995
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:Requesting worker at 0.0.0.0:62974
INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:self.control_address: localhost:62992
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.statecache:Creating state cache with size 0
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.sdk_worker:Creating insecure control channel for localhost:62992.
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.sdk_worker:Control channel established.
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.sdk_worker:Initializing SDKHarness with unbounded number of workers.
WARNING: All log messages before absl::InitializeLog() is called are written to STDERR
I0000 00:00:1754425496.285480 8399329 fork_posix.cc:77] Other threads are currently calling into gRPC, skipping fork() handlers
I0000 00:00:1754425496.575693 8399329 check_gcp_environment_no_op.cc:29] ALTS: Platforms other than Linux and Windows are not supported
INFO:apache_beam.io.gcp.bigquery:Sent BigQuery Storage API CreateReadSession request:
 data_format: AVRO
table: "projects/apache-beam-testing/datasets/charlesnguyen/tables/test"
read_options {
  selected_fields: "embedding"
  row_restriction: "id = 5"
}

Received 1 streams
data_format: 1
estimated_total_bytes_scanned: 6178480
Avro Schema:schema: "{\n    \"type\": \"record\",\n    \"name\": \"__root__\",\n    \"fields\": [\n        {\n            \"name\": \"embedding\",\n            \"type\": {\n                \"type\": \"array\",\n                \"items\": \"double\"\n            }\n        }\n    ]\n}"
.
INFO:apache_beam.io.gcp.bigquery:Started BigQuery Storage API read from stream projects/apache-beam-testing/locations/us-central1/sessions/CAISDDRMYU9NVC0zYjJYaRoCaWYaAmpx/streams/GgJpZhoCanEoAg.
I0000 00:00:1754425502.589118 8399329 fork_posix.cc:77] Other threads are currently calling into gRPC, skipping fork() handlers
I0000 00:00:1754425502.830988 8399329 check_gcp_environment_no_op.cc:29] ALTS: Platforms other than Linux and Windows are not supported
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.sdk_worker:Creating insecure state channel for localhost:62994.
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.sdk_worker:State channel established.
INFO:apache_beam.utils.subprocess_server:INFO:apache_beam.runners.worker.data_plane:Creating client data channel for localhost:62993
INFO:apache_beam.utils.subprocess_server:INFO:root:BatchElements statistics: element_count=1 batch_count=1 next_batch_size=1 timings=[]
INFO:root:{"example": [0.038988106768286936, 0.025528610286658572, -0.10666340067582115, 0.029036934163589195, -0.005526015184928034, 0.055632029892425765, 0.04430381301332253, ...

[chamikaramj]

Ah, probably the real error is being truncated by the expansion service here in that case.

I suspect the error caused the service to crash hence you just got a connection error in the retry.

[charlespnh]

Here is a minimal reproducible example of the issue:

YAML pipeline test.yaml

pipeline:
  transforms:
    - type: MyTransform
      name: MyTransform
      input: {}
      config:
        model_artifact_path: "gs://dataflow-samples/shakespeare/kinglear.txt"

providers:
  - type: pythonPackage
    config:
      packages:
        - ./dist/transform_provider-0.1.0.tar.gz
    transforms:
      MyTransform: "transform_provider.MyTransform"

Implementation of MyTransform is in transform_provider.py:

import apache_beam as beam
from apache_beam.io.filesystems import FileSystems

class MyTransform(beam.PTransform):
  def __init__(self, model_artifact_path):
    self.model_artifact_path = model_artifact_path
    self.file = FileSystems.open(self.model_artifact_path, 'r')

  def expand(self, pcoll):
    # no-op
    return (
        pcoll
    )

Building the Python distribution package with pyproject.toml below and poetry:

[tool.poetry]
name = "transform_provider"
version = "0.1.0"
description = "..."
authors = ["Your Name <you@example.com>"]
license = "Apache License 2.0"
readme = "README.md"
packages = [
    { include = "transform_provider.py" },
]


[tool.poetry.dependencies]
python = "^3.11"
apache-beam = {extras = ["gcp", "yaml"], version = "^2.66.0"}

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

Beam anomaly detection module internally uses FileSystems.open() to load the model from GCS, and this gRPC error somehow seems to be caused by FileSystems.open() usage...

[shunping]

Ack. Let me take a look

[chamikaramj]

@shunping any update on this ?
I believe #35810 is blocked by this.