LocalDagRunner can't run simple Pipeline with FileBasedExampleGen and custom parquet Executor

[jasonbrancazio]

System information

• Have I specified the code to reproduce the issue: Yes
• Environment in which the code is executed: Colab notebook, also experienced same issue on a Macbook Pro
• TensorFlow version: 2.7.0
• TFX Version: 1.5.0
• Python version: 3.7
• Python dependencies (from pip freeze output):
  • pandas (only for illustration purposes)

Describe the current behavior
The LocalDagRunner can't run an extremely simple pipeline using FilebasedExampleGen and the parquet custom executor. It reports the following error:
BaseBeamExecutor found initialized with BaseExecutorSpec. Please use BeamExecutorSpec for Beam Components instead.

Describe the expected behavior

The pipeline should run and produce examples from a parquet file.

Standalone code to reproduce the issue

This simple notebook downloads some data, converts it to a parquet file, builds a pipeline containing only a FileBasedExampleGen component with the custom parquet executor and runs it with the LocalDagRunner.
https://colab.research.google.com/drive/1ByIodrGiOkaj_qta3tMYXgkAhfuD_wlJ?usp=sharing

Thoughts

I believe this error is raised because:

• the tfx.components.example_gen.custom_executors.parquet_executor.Executor subclasses BaseExampleGenExecutor
  • https://github.com/tensorflow/tfx/blob/master/tfx/components/example_gen/custom_executors/parquet_executor.py#L58
• which subclasses BaseBeamExecutor
  • https://github.com/tensorflow/tfx/blob/master/tfx/components/example_gen/base_example_gen_executor.py#L86
• but the PythonExecutorOperator that is used by the LocalDagRunner via the Launcher is hard-coded to create a tfx.dsl.components.base.base_executor.BaseExecutor.Context, and then passes that context when initializing the parquet Executor
  • https://github.com/tensorflow/tfx/blob/master/tfx/orchestration/portable/python_executor_operator.py#L123

The error is specifically caused by this instance check (https://github.com/tensorflow/tfx/blob/master/tfx/dsl/components/base/base_beam_executor.py#L70) in the BaseBeamExecutor:

    if context:
      if isinstance(context, BaseBeamExecutor.Context):
        self._beam_pipeline_args = context.beam_pipeline_args
        self._make_beam_pipeline_fn = context.make_beam_pipeline_fn
      else:
        raise ValueError('BaseBeamExecutor found initialized with '
                         'BaseExecutorSpec. Please use BeamExecutorSpec for '
                         'Beam Components instead.')

a BaseExecutorContext is not a BaseBeamExecutor.Context, so the check fails.

[gstamatakis]

Facing the same issue, any news regarding this?

[purplechem]

Facing the same issue, any news regarding this?

[purplechem]

this simple change fixed the issue: use BeamExecutorSpec rather than ExecutorClassSpec in many wrong examples.
custom_executor_spec=executor_spec.BeamExecutorSpec(
parquet_executor.Executor))

[gaikwadrahul8]

@jasonbrancazio

We're really sorry for late response and I have reproduced the same code without error, you've to use BeamExecutorSpec() rather than ExecutorClassSpec() so I would request you to please change below code and everything will work as expected even for your reference, I have added Gist file here

Please use custom_executor_spec=executor_spec.BeamExecutorSpec( parquet_executor.Executor) instead of custom_executor_spec=executor_spec.ExecutorClassSpec( parquet_executor.Executor)

If issue still persists please let us know, we'll look into your issue and will try to resolve it as soon as possible

If your issue got resolved so please feel free to close this issue

Thank you!

[jasonbrancazio]

Thanks for the feedback @purplechem and @gaikwadrahul8 - switching to the BeamExecutorSpec indeed fixes the issue in the simple example notebook I provided.

@gaikwadrahul8, please note @purplechem's reference to "many wrong examples". I came up with my code following such examples.

These wrong examples need to be corrected in the TFX code docstrings, e.g. the "Example Usage" docstring at https://github.com/tensorflow/tfx/blob/master/tfx/components/example_gen/custom_executors/parquet_executor.py#L85, as well as in the documentation at https://www.tensorflow.org/tfx/guide/examplegen#file-based_examplegen_customization_experimental

[gaikwadrahul8]

@jasonbrancazio, @purplechem

Thank you for your confirmation and noticing the mistakes in our examples so we'll take care of those mistakes in the Examples and will correct those mistakes very soon in the TFX code docstrings as well as in the documentation, I really appreciate your efforts and valuable time

Could you please confirm if this issue is resolved for you? Please feel free to close the issue if it is resolved ?

Thank you!

[google-ml-butler]

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