You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/06/04 21:55:17 UTC
[GitHub] [beam] damccorm opened a new issue, #21229: Error running Top transform in Dataflow
damccorm opened a new issue, #21229:
URL: https://github.com/apache/beam/issues/21229
When running the following transform in Dataflow (the problem does not happen with the direct running)... I is a streaming pipeline where I am using a SlidingWindow.
```
beam.combiners.Top.Of(n=10, key=lambda item: item[1]).without_defaults()
```
I am getting an error
```
Error message from worker: generic::unknown: Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 284, in _execute response = task() File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 357, in <lambda> lambda: self.create_worker().do_instruction(request), request) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 602, in do_instruction getattr(request, request_type), request.instruction_id) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 639, in process_bundle bundle_processor.process_bundle(instruction_id)) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 994, in process_bundle element.data) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 222, in process_encoded self.output(decoded_value) File "apache_beam/runners/worker/operations.py",
line 351, in apache_beam.runners.worker.operations.Operation.output File "apache_beam/runners/worker/operations.py",
line 353, in apache_beam.runners.worker.operations.Operation.output File "apache_beam/runners/worker/operations.py",
line 215, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive File "apache_beam/runners/worker/operations.py",
line 921, in apache_beam.runners.worker.operations.CombineOperation.process File "apache_beam/runners/worker/operations.py",
line 925, in apache_beam.runners.worker.operations.CombineOperation.process File "/usr/local/lib/python3.7/site-packages/apache_beam/transforms/combiners.py",
line 835, in extract_only return self.combine_fn.extract_output(accumulator) File "/usr/local/lib/python3.7/site-packages/apache_beam/transforms/combiners.py",
line 502, in extract_output heap.sort(reverse=True) File "apache_beam/transforms/cy_combiners.py", line
389, in apache_beam.transforms.cy_combiners.ComparableValue.__lt__ AssertionError passed through: ==>
dist_proc/dax/workflow/worker/fnapi_service_impl.cc:644
```
`What might be the cause of this issue?`
Imported from Jira [BEAM-12847](https://issues.apache.org/jira/browse/BEAM-12847). Original Jira may contain additional context.
Reported by: mirene.
--
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: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Error running Top transform in Dataflow [beam]
Posted by "jrecuerda (via GitHub)" <gi...@apache.org>.
jrecuerda commented on issue #21229:
URL: https://github.com/apache/beam/issues/21229#issuecomment-2432471288
Experiencing the same issue when running a streaming dataflow. I am using apache-beam==2.60.0
```
Error message from worker: generic::unknown: Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 311, in _execute
response = task()
^^^^^^
File "/usr/local/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 386, in <lambda>
lambda: self.create_worker().do_instruction(request), request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 656, in do_instruction
return getattr(self, request_type)(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 694, in process_bundle
bundle_processor.process_bundle(instruction_id))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1119, in process_bundle
input_op_by_transform_id[element.transform_id].process_encoded(
File "/usr/local/lib/python3.11/site-packages/apache_beam/runners/worker/bundle_processor.py", line 237, in process_encoded
self.output(decoded_value)
File "apache_beam/runners/worker/operations.py", line 567, in apache_beam.runners.worker.operations.Operation.output
File "apache_beam/runners/worker/operations.py", line 569, in apache_beam.runners.worker.operations.Operation.output
File "apache_beam/runners/worker/operations.py", line 260, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
File "apache_beam/runners/worker/operations.py", line 263, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
File "apache_beam/runners/worker/operations.py", line 1159, in apache_beam.runners.worker.operations.CombineOperation.process
File "apache_beam/runners/worker/operations.py", line 1163, in apache_beam.runners.worker.operations.CombineOperation.process
File "/usr/local/lib/python3.11/site-packages/apache_beam/transforms/combiners.py", line 921, in extract_only
return self.combine_fn.extract_output(accumulator)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/apache_beam/transforms/combiners.py", line 541, in extract_output
heap.sort(reverse=True)
File "apache_beam/transforms/cy_combiners.py", line 419, in apache_beam.transforms.cy_combiners.ComparableValue.__lt__
AssertionError
passed through:
==>
dist_proc/dax/workflow/worker/fnapi_service_impl.cc:1191"
```
Code to reproduce it:
```
class PrintFn(beam.DoFn):
"""write one file per window/key."""
def process(self, elements, window=beam.DoFn.WindowParam):
package = window.start.to_utc_datetime().strftime('%Y%m%dT%H%M')
logging.info('JRH_TEST_PACKAGE:%s:%s', package, ';'.join(elements))
def run():
"""The main function which creates the pipeline and runs it."""
pipeline_options = PipelineOptions()
p = beam.Pipeline(options=pipeline_options)
feed_options = pipeline_options.view_as(FeedOptions)
pipeline_options = PipelineOptions()
with beam.Pipeline(options=pipeline_options) as p:
top_combiner = beam.combiners.Top.Of(n=2, key=len)
top_combiner.has_defaults = False
_ = (p
| beam.io.ReadFromPubSub(subscription=feed_options.input_subscription)
| beam.WindowInto(
beam.window.FixedWindows(60 * 5),
accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING)
| beam.Map(lambda element: element.decode())
| beam.FlatMap(lambda elements: elements.split())
| top_combiner
| beam.ParDo(PrintFn())
)
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Vidyats1 commented on issue #21229: Error running Top transform in Dataflow
Posted by GitBox <gi...@apache.org>.
Vidyats1 commented on issue #21229:
URL: https://github.com/apache/beam/issues/21229#issuecomment-1176848520
Even I am getting the same error while running below code in dataflow. I am using apache-beam==2.40.0
`beam.CombineGlobally(beam.combiners.TopCombineFn(5,lambda a: a['count'] )).without_defaults()`
```
severity: "ERROR"
textPayload: "Error message from worker: generic::unknown: Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 284, in _execute
response = task()
File "/usr/local/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 357, in <lambda>
lambda: self.create_worker().do_instruction(request), request)
File "/usr/local/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 597, in do_instruction
return getattr(self, request_type)(
File "/usr/local/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 635, in process_bundle
bundle_processor.process_bundle(instruction_id))
File "/usr/local/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1003, in process_bundle
input_op_by_transform_id[element.transform_id].process_encoded(
File "/usr/local/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 227, in process_encoded
self.output(decoded_value)
File "apache_beam/runners/worker/operations.py", line 526, in apache_beam.runners.worker.operations.Operation.output
File "apache_beam/runners/worker/operations.py", line 528, in apache_beam.runners.worker.operations.Operation.output
File "apache_beam/runners/worker/operations.py", line 237, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
File "apache_beam/runners/worker/operations.py", line 240, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
File "apache_beam/runners/worker/operations.py", line 1119, in apache_beam.runners.worker.operations.CombineOperation.process
File "apache_beam/runners/worker/operations.py", line 1123, in apache_beam.runners.worker.operations.CombineOperation.process
File "/usr/local/lib/python3.9/site-packages/apache_beam/transforms/combiners.py", line 889, in extract_only
return self.combine_fn.extract_output(accumulator)
File "/usr/local/lib/python3.9/site-packages/apache_beam/transforms/combiners.py", line 526, in extract_output
heap.sort(reverse=True)
File "apache_beam/transforms/cy_combiners.py", line 421, in apache_beam.transforms.cy_combiners.ComparableValue.__lt__
AssertionError
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org