apache / apache/beam

[Bug]: ValueError: Invalid DisplayDataItem when using AsSingleton for side input (need better parameter validation before it hits this)

Open
#25,860 3 comments 0 reactions 0 assignees View on GitHub
bug P2 python
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### What happened?

Using a side input in the form in WriteToText with Dataflow as the runner causes an error.

` File ".venv\lib\site-packages\apache_beam\runners\dataflow\dataflow_runner.py", line 877, in run_ParDo
step = self._add_step(
File ".venv\lib\site-packages\apache_beam\runners\dataflow\dataflow_runner.py", line 652, in _add_step
[
File ".venv\lib\site-packages\apache_beam\runners\dataflow\dataflow_runner.py", line 653, in
item.get_dict()
File ".venv\lib\site-packages\apache_beam\transforms\display.py", line 370, in get_dict
self.is_valid()
File ".venv\lib\site-packages\apache_beam\transforms\display.py", line 336, in is_valid
raise ValueError(
ValueError: Invalid DisplayDataItem. Value is of an unsupported type.`

The pipeline works fine when running locally but fails when using a Dataflow runner. Tested with all the different `beam.pvalue.As...` too and it still happens.

SDK version with the error: 2.46.0
Python version used: 3.9.13

Code of the whole pipeline:

```
import argparse
import json
import logging
from typing import Any, Dict, List

import apache_beam as beam
import apache_beam.io.fileio
from apache_beam.options.pipeline_options import PipelineOptions

def is_valid(json_file):
"""
This function is used as a filter.

Args:
json_file: json file corresponding to one episode.

Returns:
True if the episode is valid, False otherwise.
"""
with json_file.open() as f:
first_line = f.readline()
first_line_json = json.loads(first_line)
return first_line_json['username'] == "username"

def run(
input_raw_data: str,
output_path: str,
beam_args: List[str] = None,
) -> None:
"""Build and run the pipeline."""
options = PipelineOptions(beam_args, save_main_session=True, streaming=False)

with beam.Pipeline(options=options) as pipeline:
filtered_files = (
pipeline
| 'Match file paths' >> beam.io.fileio.MatchFiles(input_raw_data)
| 'Read file paths' >> beam.io.fileio.ReadMatches()
| 'Filter correct episodes' >> beam.Filter(is_valid)
)
file_count = (
filtered_files
| 'Count files' >> beam.combiners.Count.Globally()
)
#removing the file_count_integer as a side input fixes the error.
file_count_integer = beam.pvalue.AsSingleton(file_count)
(
filtered_files
| 'Read files' >> beam.Map(lambda x: x.read_utf8())
| 'Write to files' >> beam.io.WriteToText(
output_path,
file_name_suffix='.json',
num_shards=file_count_integer)
)

if __name__ == "__main__":
logging.getLogger().setLevel(logging.INFO)

parser = argparse.ArgumentParser()
parser.add_argument(
"--output-path",
required=True,
help="Name of the folder to output the cleaned data files. ",
)
parser.add_argument(
"--input-raw-data",
required=True,
help="Name of the folder containing the raw data files." ,
)
args, beam_args = parser.parse_known_args()

run(
input_raw_data=args.input_raw_data,
output_path=args.output_path,
beam_args=beam_args,
)
```

Command used to run:

```
python -m script.py --region europe-west1 --input gs://path-to-files/*.json --output gs://path-to-utputs/results/outputs --runner DataflowRunner --project google_project --temp_location gs://temp_bucket/tmp/
```

### Issue Priority

Priority: 1 (data loss / total loss of function)

### Issue Components

- [X] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] 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

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.