[Bug]: TypeError: Argument 'input_args' has incorrect type (expected list, got tuple)

[tuzhucheng]

What happened?

My colleague has a simple piece of code that reads some lines from a text file, and then emits the lines to a different file. However, we see the error TypeError: Argument 'input_args' has incorrect type (expected list, got tuple).

I am using apache-beam==2.51.0.

Here is some minimal code that reproduced the bug:

Suppose /tmp/examples.jsonl is a file with the following content:

1
2

Suppose my code is beam_bug_minimal_repro.py:

import apache_beam as beam
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


class GenerateFnExampleFn(beam.DoFn):
    def process(self, fn_jsonl: str):
        with open(fn_jsonl) as fp:
            for line in fp:
                yield line


pipeline_options = PipelineOptions([])
with beam.Pipeline(options=pipeline_options) as pipeline:
    # Read the text file[pattern] into a PCollection.
    shard_indices = (
        # pipeline | ReadFromTextWithFilename(input)
            pipeline | beam.Create(["/tmp/examples.jsonl"])
    )
    # Datasets which feed file names.
    examples = shard_indices | "GenerateExamples" >> beam.ParDo(
        GenerateFnExampleFn()
    )
    examples = examples | "Reshuffle" >> beam.Reshuffle()
    # Write the output using a "Write" transform that has side effects.
    # pylint: disable=expression-not-assigned
    examples | "Write" >> WriteToText("/tmp/test_out.txt")

Running the script gives the following error:

$ python beam_bug_minimal_repro.py
Traceback (most recent call last):
  File "beam_bug_minimal_repro.py", line 27, in <module>
    examples | "Write" >> WriteToText("/tmp/test_out.txt")
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/pipeline.py", line 600, in __exit__
    self.result = self.run()
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/pipeline.py", line 577, in run
    return self.runner.run_pipeline(self, self._options)
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/direct/direct_runner.py", line 128, in run_pipeline
    return runner.run_pipeline(pipeline, options)
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 202, in run_pipeline
    self._latest_run_result = self.run_via_runner_api(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 224, in run_via_runner_api
    return self.run_stages(stage_context, stages)
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 455, in run_stages
    bundle_results = self._execute_bundle(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 783, in _execute_bundle
    self._run_bundle(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1020, in _run_bundle
    result, splits = bundle_manager.process_bundle(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1356, in process_bundle
    result_future = self._worker_handler.control_conn.push(process_bundle_req)
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/worker_handlers.py", line 379, in push
    response = self.worker.do_instruction(request)
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 625, in do_instruction
    return getattr(self, request_type)(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 656, in process_bundle
    bundle_processor = self.bundle_processor_cache.get(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 487, in get
    processor = bundle_processor.BundleProcessor(
  File "/Users/michaeltu/miniforge3/envs/myenv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 914, in __init__
    op.setup(self.data_sampler)
  File "apache_beam/runners/worker/operations.py", line 877, in apache_beam.runners.worker.operations.DoOperation.setup
  File "apache_beam/runners/worker/operations.py", line 915, in apache_beam.runners.worker.operations.DoOperation.setup
  File "apache_beam/runners/common.py", line 1421, in apache_beam.runners.common.DoFnRunner.__init__
  File "apache_beam/runners/common.py", line 502, in apache_beam.runners.common.DoFnInvoker.create_invoker
  File "apache_beam/runners/common.py", line 771, in apache_beam.runners.common.PerWindowInvoker.__init__
TypeError: Argument 'input_args' has incorrect type (expected list, got tuple)

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: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[AnandInguva]

https://github.com/apache/beam/pull/28385/files#r1321953437. This PR was merged after 2.51.0 release and should be available in 2.52.0. Let me check with the 2.52.0 RC.

I got the similar error, when I was testing code using apache beam wheel which has cython extensions available.

[AnandInguva]

I can't reproduce on my M1 MacOS. The pipeline is passing with 2.51.0. Can you try with the 2.52.0 RC3 - https://pypi.org/project/apache-beam/2.52.0rc3/ using pip install apache-beam==2.52.0rc3 and see if it solves your issue?

[tuzhucheng]

Hi @AnandInguva, this seems to be resolved with apache-beam==2.52.0rc3. Thanks!