tuzhucheng opened a new issue, #29361:
URL: https://github.com/apache/beam/issues/29361
### 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:
```txt
1
2
```
Suppose my code is `beam_bug_minimal_repro.py`:
```python
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
- [X] 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
