https://flink.apache.org/ logo
Join Slack
Powered by
Hi Team, I am trying to run PyFlink 1.17 on Mac Wh...
# troubleshooting
n
Hi Team, I am trying to run PyFlink 1.17 on Mac When using StreamEnvironment it throws me an error
Copy code
TypeError: Argument 'input_stream' has incorrect type (expected pyflink.fn_execution.stream_fast.LengthPrefixInputStream, got BeamInputStream)
It comes when when I add a streaming source to the environment. Any work around the error?
d
@Nitin Bansal This is a known issue which happens on Mac M1. It will be fixed in 1.17.1. For now, you could try to work around it by reinstalling PyFlink with the following commands: • pip install cython==0.29.24 • brew install gcc • pip uninstall apache-flink • pip uninstall apache-beam • pip install apache-flink See https://issues.apache.org/jira/browse/FLINK-28786 for more details.
👍 1
n
Thanks a lot for fast resolution
Copy code
Traceback (most recent call last):
  File "/opt/homebrew/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 287, in _execute
    response = task()
  File "/opt/homebrew/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 360, in <lambda>
    lambda: self.create_worker().do_instruction(request), request)
  File "/opt/homebrew/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 596, in do_instruction
    return getattr(self, request_type)(
  File "/opt/homebrew/lib/python3.10/site-packages/apache_beam/runners/worker/sdk_worker.py", line 634, in process_bundle
    bundle_processor.process_bundle(instruction_id))
  File "/opt/homebrew/lib/python3.10/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 "/opt/homebrew/lib/python3.10/site-packages/apache_beam/runners/worker/bundle_processor.py", line 225, in process_encoded
    decoded_value = self.windowed_coder_impl.decode_from_stream(
  File "/opt/homebrew/lib/python3.10/site-packages/apache_beam/coders/coder_impl.py", line 1520, in decode_from_stream
    value = self._value_coder.decode_from_stream(in_stream, nested)
  File "/opt/homebrew/lib/python3.10/site-packages/pyflink/fn_execution/beam/beam_coder_impl_slow.py", line 34, in decode_from_stream
    return self._value_coder.decode_from_stream(in_stream, nested)
  File "/opt/homebrew/lib/python3.10/site-packages/pyflink/fn_execution/beam/beam_coder_impl_slow.py", line 58, in decode_from_stream
    return self._value_coder.decode_from_stream(data_input_stream)
TypeError: Argument 'input_stream' has incorrect type (expected pyflink.fn_execution.stream_fast.LengthPrefixInputStream, got BeamInputStream)
This error still persists
https://github.com/apache/flink/commit/52c9742eed7128284278b07c40785bf1c4e30139#diff-caba223baedad8d2bc0c3c2d955[…]4cc3ca6677870cd45f6289af56aa70L22 This will also return True if someone sets any value in environment variable, right? Also even after setting the environment variable it's going in slow implementation.
Copy code
ValueError: apache_beam.runners.worker.operations.Operation size changed, may indicate binary incompatibility. Expected 136 from C header, got 32 from PyObject
This is coming while importing 
from pyflink.fn_execution.beam import beam_operations_fast
Copy code
ModuleNotFoundError: No module named 'apache_beam.coders.stream'
while 
from pyflink.fn_execution.beam import beam_stream_fast
Copy code
ImportError: cannot import name 'window_aggregate_fast' from 'pyflink.fn_execution.beam'
d
@Nitin Bansal Could you check whether Beam’s fast by checking whether there are files such as 
python3.7/site-packages/apache_beam/coders/coder_impl.cpython-37m-darwin.so
?
t
For anyone finding this later, you have to make sure to download a fresh version of flink, so use 
--no-cache-dir
when installing with pip.
4 Views
Open in Slack
PreviousNext
Powered by