Cannot successfully execute the TFX examples ranking pipeline with LocalDagRunner

[etsyzlong]

System information

• Have I specified the code to reproduce the issue (Yes, No): No, TFX example code
• Environment in which the code is executed (e.g., Local(Linux/MacOS/Windows), Interactive Notebook, Google Cloud, etc): Linux
• TensorFlow version: 2.5.0
• TFX Version: 0.30.1
• Python version: 3.7.5
• Python dependencies (from pip freeze output): Relevant output:

  struct2tensor==0.32.0
  tensorflow==2.5.0
  tensorflow-addons==0.13.0
  tensorflow-data-validation==0.30.0
  tensorflow-datasets==4.3.0
  tensorflow-estimator==2.5.0
  tensorflow-hub==0.9.0
  tensorflow-metadata==1.1.0
  tensorflow-model-analysis==0.30.0
  tensorflow-model-optimization==0.6.0
  tensorflow-ranking==0.4.0
  tensorflow-serving-api==2.4.1
  tensorflow-transform==0.30.0
  tfx==0.30.1
  tfx-bsl==0.30.0

Describe the current behavior Running python ranking_pipeline.py from https://github.com/tensorflow/tfx/blob/master/tfx/examples/ranking directory yields the following error:

FileNotFoundError: Op type not registered 'DecodeProtoSparseV3' in binary running on <machine_name>.

Added import struct2tensor to line 20 of the ranking_pipeline.py file did not resolve the issue.

Describe the expected behavior

Running python ranking_pipeline.py should run the ranking pipeline locally and successfully.

Standalone code to reproduce the issue

from tfx repo root:

cd tfx/examples/ranking
python ranking_pipeline.py

[chongkong]

Thanks @etsyzlong. This is a bug and we're tracking it. This happens because import struct2tensor does not happen inside the worker thread/subprocesses. For a quick fix, you may run the example with the following beam arguments:

beam_pipeline_args = [
    '--direct_running_mode=in_memory',
    '--direct_num_workers=1',
]

which basically runs the pipeline in the current thread. We are working on a fix for the multi thread/process environment, but is still on discussion and might take some time.

[etsyzlong]

Thank you @chongkong, the pipeline is working with the beam configurations.

[eunicevin]

If you need to run your pipeline in a multi-process environment like kubeflow or airflow, there is a workaround. Modify to import struct2tensor from tfx-bsl, package it and dockerization.

[UsharaniPagadala]

@etsyzlong

Closing this issue as it is resolved.Please feel free to reopen if this still exists.Thanks

[google-ml-butler[bot]]

Are you satisfied with the resolution of your issue? Yes No