Hi all Are there any channels for python users I am trying t Apache Flink #troubleshooting

Hi all, Are there any channels for python users? ...

Christos Hadjinikolis

10/22/2023, 10:31 PM

Hi all, Are there any channels for python users? I am trying to setup a local test. Looks like this:

Copy code

import os

import pytest
from pyflink.datastream import RuntimeExecutionMode, StreamExecutionEnvironment

from src.dataflows.pipelines import main_pipeline
from src.operators.transformers import PrintSinkFunction
from tests.config.constants import SAMPLE_NEXT_DESTINATION_MESSAGES, SAMPLE_TAXI_MESSAGES
from tests.utils.loaders import load_next_destination_messages, load_taxi_details_messages


@pytest.mark.end_to_end
def test_process_taxis_stream(taxi_details_serde, next_destination_serde):
    # Given
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
    env.set_parallelism(1)
    env.set_python_requirements(requirements_file_path=os.path.join(os.path.dirname(os.path.realpath(__file__)), "../requirements.txt"))

    # Load sample taxi messages
    sample_taxi_messages = load_taxi_details_messages(SAMPLE_TAXI_MESSAGES)
    taxis_stream = env.from_collection(sample_taxi_messages)

    # Load sample next destination messages
    sample_next_destination_messages = load_next_destination_messages(SAMPLE_NEXT_DESTINATION_MESSAGES)
    next_destinations_stream = env.from_collection(sample_next_destination_messages)

    # When
    updated_taxis_stream = main_pipeline(taxis_stream, next_destinations_stream, taxi_details_serde, next_destination_serde, verbose=True)

    # Then
    updated_taxis_stream.process(PrintSinkFunction())

    env.execute("Taxis Stream Processing (End-to-End Test)")
    assert updated_taxis_stream is not None

Practically, I would be reading the inputs from two kafka topics... I am replacing them with local json data that I am parsing to generate the two lists

sample_next_destination_messages

and

next_destinations_stream

. The point is, I am trying to test the main logic (which include complicated stateful processing) locally. Every time I run it though I get:

Copy code

>       env.execute("Taxis Stream Processing (End-to-End Test)")

test_main_pipeline.py:62: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../.venv/lib/python3.10/site-packages/pyflink/datastream/stream_execution_environment.py:763: in execute
    j_stream_graph = self._generate_stream_graph(clear_transformations=True, job_name=job_name)
../.venv/lib/python3.10/site-packages/pyflink/datastream/stream_execution_environment.py:1004: in _generate_stream_graph
    JPythonConfigUtil.configPythonOperator(self._j_stream_execution_environment)
../.venv/lib/python3.10/site-packages/py4j/java_gateway.py:1322: in __call__
    return_value = get_return_value(
../.venv/lib/python3.10/site-packages/pyflink/util/exceptions.py:146: in deco
    return f(*a, **kw)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

answer = 'xro224'
gateway_client = <py4j.java_gateway.GatewayClient object at 0x1036da410>
target_id = 'z:org.apache.flink.python.util.PythonConfigUtil'
name = 'configPythonOperator'

    def get_return_value(answer, gateway_client, target_id=None, name=None):
        """Converts an answer received from the Java gateway into a Python object.
    
        For example, string representation of integers are converted to Python
        integer, string representation of objects are converted to JavaObject
        instances, etc.
    
        :param answer: the string returned by the Java gateway
        :param gateway_client: the gateway client used to communicate with the Java
            Gateway. Only necessary if the answer is a reference (e.g., object,
            list, map)
        :param target_id: the name of the object from which the answer comes from
            (e.g., *object1* in `object1.hello()`). Optional.
        :param name: the name of the member from which the answer comes from
            (e.g., *hello* in `object1.hello()`). Optional.
        """
        if is_error(answer)[0]:
            if len(answer) > 1:
                type = answer[1]
                value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
                if answer[1] == REFERENCE_TYPE:
>                   raise Py4JJavaError(
                        "An error occurred while calling {0}{1}{2}.\n".
                        format(target_id, ".", name), value)
E                   py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.flink.python.util.PythonConfigUtil.configPythonOperator.
E                   : java.lang.reflect.InaccessibleObjectException: Unable to make field final java.util.Map java.util.Collections$UnmodifiableMap.m accessible: module java.base does not "opens java.util" to unnamed module @71aa8854
E                   	at java.base/java.lang.reflect.AccessibleObject.throwInaccessibleObjectException(AccessibleObject.java:391)
E                   	at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:367)
E                   	at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:315)
E                   	at java.base/java.lang.reflect.Field.checkCanSetAccessible(Field.java:183)
E                   	at java.base/java.lang.reflect.Field.setAccessible(Field.java:177)
E                   	at org.apache.flink.python.util.PythonConfigUtil.registerPythonBroadcastTransformationTranslator(PythonConfigUtil.java:357)
E                   	at org.apache.flink.python.util.PythonConfigUtil.configPythonOperator(PythonConfigUtil.java:101)
E                   	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
E                   	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
E                   	at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
E                   	at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
E                   	at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
E                   	at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
E                   	at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
E                   	at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
E                   	at java.base/java.lang.Thread.run(Thread.java:1583)

../.venv/lib/python3.10/site-packages/py4j/protocol.py:326: Py4JJavaError

Christos Hadjinikolis

10/22/2023, 10:33 PM

I am using

python==3.10.10

and

java==11.0.11

with pyFlink -->

Copy code

apache-flink = "1.17.1"

Christos Hadjinikolis

10/22/2023, 10:33 PM

Is it possible to run this locally ?

Jaehyeon Kim

10/23/2023, 1:55 AM

It is not easy to check what's wrong because your source functions are imported from other places. I recently tried some unit testing on Pyflink data streams and it works fine. It is similar to Pyspark unit teseting. It can be found in https://jaehyeon.me/blog/2023-10-19-build-pyflink-apps/

Christos Hadjinikolis

10/23/2023, 7:58 AM

Unit testing is fine with me too. This is an end-to-end test though. I am needed to run

env.execute()

Christos Hadjinikolis

10/23/2023, 7:59 AM

I think I need to run flink locally for this test to run

M C (Mac)

10/23/2023, 10:21 AM

Naive answer - usually I get that error for incorrect Java versions. Can see you have the version specified, is that the only referenceable Java version on the host OS?

Christos Hadjinikolis

10/23/2023, 10:31 AM

No there are many. But I am using jenv to make sure I am using the correct one.

Christos Hadjinikolis

10/23/2023, 10:31 AM

I have also configured pyCharm run configuration to use the correct Java Home 🏠

👌 1

Siddhesh Kalgaonkar

10/25/2023, 5:55 AM

@Christos Hadjinikolis for

env.execute

to run as expected you need to setup some sink too. I came across one of the posts a few months back where the user tried something similar and was recommended the same thing.

Siddhesh Kalgaonkar

10/25/2023, 5:57 AM

Alternative: If you want to skip that, switch to

Python unittests

library. I am not familiar with

pytest

. I literally tested everything I implemented by mocking stuffs using

unittest

lib. Also for e2e, you can use that. Thats what I did.

Christos Hadjinikolis

10/27/2023, 10:39 AM

Thanks for the input but I am facing issues long before I get to running anything at all..

Christos Hadjinikolis

10/27/2023, 10:39 AM

Basically, I'm running into an issue while trying to execute a Flink job with Python dependencies. When Flink's Python environment attempts to install the dependencies, it uses the

--install-option

argument with the

pip install

command, which results in an error due to this option being unsupported in recent

pip

versions. Error:

Copy code

no such option: --install-option

Command that fails:

Copy code

python -m pip install --ignore-installed -r /path/to/requirements.txt --install-option --prefix=/path/to/python-requirements

Setup: - Flink version: 1.17.1 - Python version: 3.10 - pip version: 23.3.1 I've verified that my environment uses compatible versions of Java, Python, and pip as per Flink's requirements. Has anyone else encountered this issue? Are there any known workarounds or fixes available? Any guidance would be greatly appreciated. Thank you!

2 Views

Open in Slack

Previous Next