tuzhucheng opened a new issue, #26330:
URL: https://github.com/apache/beam/issues/26330
### What happened?
I am using Apache Beam with GCP Dataflow. I am just doing a straightforward
`beam.Map`, but encountered the following stacktrace. I've encounter this on
various versions including as 2.38.0, 2.42.0, 2.43.0 using both dill (both
0.3.1.1 and 0.3.6) and cloudpickle. On 2.46.0 when using cloudpickle with
`--save_main_session` enabled, I get a different error with `ModuleNotFound`,
cannot find module `tensorflow_datasets`.
```python
File
"/opt/venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 1198, in create_operation
return creator(self, transform_id, transform_proto, payload, consumers)
File
"/opt/venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 1545, in create_par_do
return _create_pardo_operation(
File
"/opt/venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 1588, in _create_pardo_operation
dofn_data = pickler.loads(serialized_fn)
File
"/opt/venv/lib/python3.9/site-packages/apache_beam/internal/pickler.py", line
51, in loads
return desired_pickle_lib.loads(
File
"/opt/venv/lib/python3.9/site-packages/apache_beam/internal/dill_pickler.py",
line 289, in loads
return dill.loads(s)
File "/opt/venv/lib/python3.9/site-packages/dill/_dill.py", line 275, in
loads
return load(file, ignore, **kwds)
File "/opt/venv/lib/python3.9/site-packages/dill/_dill.py", line 270, in
load
return Unpickler(file, ignore=ignore, **kwds).load()
File "/opt/venv/lib/python3.9/site-packages/dill/_dill.py", line 472, in
load
obj = StockUnpickler.load(self)
AttributeError: can't set attribute
```
My mapper is like this:
```python
def parse_example(serialized_example: str) -> Tuple[str, Dict[str, Any]]:
import json
import uuid
example = json.loads(serialized_example)
uid = example.get("uuid", str(uuid.uuid4()))
return uid, {
"uuid": uid,
"text": example.get("text", "") or "",
"label": float(example["label"]) if ("label" in example and
example["label"] is not None) else -1,
}
```
Feature signature is like this:
```python
import tensorflow_datasets as tfds
....
{
"uuid": tfds.features.Text(),
"label": tfds.features.Scalar(dtype=tf.float32),
}
```
Beam pipeline is like this where `mapping_fn` is `beam.Map(parse_example)`:
```python
class TfdsBuilder(tfds.core.GeneratorBasedBuilder):
def _generate_examples(self, split_dir: str) ->
tfds.core.split_builder.SplitGenerator:
return (
beam.Create(tf.io.gfile.glob(split_dir))
| beam.io.ReadAllFromText()
| mapping_fn
)
```
### 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 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]
