https://flink.apache.org/ logo
Join Slack
Powered by
Does this error mean anything to anyone? This is o...
# troubleshooting
n
Does this error mean anything to anyone? This is on flink 1.17.0 after running a message through a custom process function
Copy code
2023-05-02 16:31:49,152 ERROR /usr/local/lib/python3.8/dist-packages/pyflink/fn_execution/beam/beam_sdk_worker_main.py:35 [] - Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 287, in _execute
    response = task()
  File "/usr/local/lib/python3.8/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 360, in <lambda>
    lambda: self.create_worker().do_instruction(request), request)
  File "/usr/local/lib/python3.8/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 596, in do_instruction
    return getattr(self, request_type)(
  File "/usr/local/lib/python3.8/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 634, in process_bundle
    bundle_processor.process_bundle(instruction_id))
  File "/usr/local/lib/python3.8/dist-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.8/dist-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 "pyflink/fn_execution/beam/beam_operations_fast.pyx", line 169, in pyflink.fn_execution.beam.beam_operations_fast.FunctionOperation.process
  File "pyflink/fn_execution/beam/beam_operations_fast.pyx", line 194, in pyflink.fn_execution.beam.beam_operations_fast.FunctionOperation.process
  File "pyflink/fn_execution/beam/beam_operations_fast.pyx", line 104, in pyflink.fn_execution.beam.beam_operations_fast.IntermediateOutputProcessor.process_outputs
  File "pyflink/fn_execution/beam/beam_operations_fast.pyx", line 169, in pyflink.fn_execution.beam.beam_operations_fast.FunctionOperation.process
  File "pyflink/fn_execution/beam/beam_operations_fast.pyx", line 194, in pyflink.fn_execution.beam.beam_operations_fast.FunctionOperation.process
  File "pyflink/fn_execution/beam/beam_operations_fast.pyx", line 92, in pyflink.fn_execution.beam.beam_operations_fast.NetworkOutputProcessor.process_outputs
  File "pyflink/fn_execution/beam/beam_coder_impl_fast.pyx", line 101, in pyflink.fn_execution.beam.beam_coder_impl_fast.FlinkLengthPrefixCoderBeamWrapper.encode_to_stream
  File "pyflink/fn_execution/coder_impl_fast.pyx", line 273, in pyflink.fn_execution.coder_impl_fast.IterableCoderImpl.encode_to_stream
  File "pyflink/fn_execution/coder_impl_fast.pyx", line 401, in pyflink.fn_execution.coder_impl_fast.RowCoderImpl.encode_to_stream
  File "pyflink/fn_execution/coder_impl_fast.pyx", line 391, in pyflink.fn_execution.coder_impl_fast.RowCoderImpl.encode_to_stream
AttributeError: 'dict' object has no attribute 'get_fields_by_names'
It's way too in the weeds of internal implementation for me to understand what went wrong.
Confirmed my container is running with flink 1.17.0 and beam 2.43.0
My process function runs fine in unit tests which is strange
I figured this out
I was outputting a 
dict
when I should have been outputting a 
Row
based on the type info I told flink about.
When it went into encoding, things were unhappy with this error reported.
a
yeah, I’ve seen that a lot 🙂
FWIW, we have a convert_dict_to_rows function that will automate this (if you have the RowTypeInfo):
🙌 1
2 Views
Open in Slack
PreviousNext
Powered by