mahrsee1997 opened a new issue, #24665:
URL: https://github.com/apache/beam/issues/24665
### What happened?
Code:
```
request
| 'Fetch' >> beam.ParDo(FetchData())
| 'Add timestamps' >> beam.Map(lambda x:
(apache_beam.utils.timestamp.Timestamp.now(), x))
| 'GroupByKey' >> beam.GroupByKey()
| 'Download' >> beam.ParDo(DownloadData())
```
Error:
```
WARNING 22-12-14 18:38:47 apache_beam.coders.coder_impl: Using fallback
deterministic coder for type '<class 'apache_beam.utils.timestamp.Timestamp'>'
in 'Fetch data/GroupByKey'.
Traceback (most recent call last):
File "apache_beam/runners/common.py", line 1417, in
apache_beam.runners.common.DoFnRunner.process
File "apache_beam/runners/common.py", line 623, in
apache_beam.runners.common.SimpleInvoker.invoke_process
File "apache_beam/runners/common.py", line 1581, in
apache_beam.runners.common._OutputHandler.handle_process_outputs
File "apache_beam/runners/common.py", line 1694, in
apache_beam.runners.common._OutputHandler._write_value_to_tag
File "apache_beam/runners/worker/operations.py", line 239, in
apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
File "apache_beam/runners/worker/operations.py", line 198, in
apache_beam.runners.worker.operations.ConsumerSet.update_counters_start
File "apache_beam/runners/worker/opcounters.py", line 215, in
apache_beam.runners.worker.opcounters.OperationCounters.update_from
File "apache_beam/runners/worker/opcounters.py", line 267, in
apache_beam.runners.worker.opcounters.OperationCounters.do_sample
File "apache_beam/coders/coder_impl.py", line 1455, in
apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
File "apache_beam/coders/coder_impl.py", line 1466, in
apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
File "apache_beam/coders/coder_impl.py", line 1015, in
apache_beam.coders.coder_impl.AbstractComponentCoderImpl.get_estimated_size_and_observables
File "apache_beam/coders/coder_impl.py", line 207, in
apache_beam.coders.coder_impl.CoderImpl.get_estimated_size_and_observables
File "apache_beam/coders/coder_impl.py", line 1544, in
apache_beam.coders.coder_impl.LengthPrefixCoderImpl.estimate_size
File "apache_beam/coders/coder_impl.py", line 246, in
apache_beam.coders.coder_impl.StreamCoderImpl.estimate_size
File "apache_beam/coders/coder_impl.py", line 457, in
apache_beam.coders.coder_impl.FastPrimitivesCoderImpl.encode_to_stream
File "apache_beam/coders/coder_impl.py", line 518, in
apache_beam.coders.coder_impl.FastPrimitivesCoderImpl.encode_special_deterministic
TypeError: Unable to deterministically encode 'Timestamp(1671043127.434635)'
of type '<class 'apache_beam.utils.timestamp.Timestamp'>', please provide a
type hint for the input of 'Fetch data/GroupByKey
```
Objective for doing the same -- I need to break the fusion b/w two ParDo
i.e. FetchData() & DownloadData() so I thought to create element key as current
TimeStamp & with the hope that due to this `GroupByKey()` will not be waiting
for all the elements to get processed first (of the above steps) as current
TimeStamp will be a unique key so no sense of waiting at GroupByKey() step. And
it will pass on the element to next pipeline step i.e. DownloadData()
immediately -- which in turn should run in separate worker.
Can't add simple Reshuffle()/GroupBy() b/w two ParDo as I don't want to wait
for FetchData() step to finish first & then begin with downloading data in
DownloadData() step. **I want to run both of these steps in parallel.** Any
other better way for doing the same ? Or this will work after resolving the
error ?
### 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
- [X] 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]
