Hey All trying to write unit tests for pyflink I can t seem Apache Flink #troubleshooting

Hey All, trying to write unit tests for pyflink. ...

David Bryson

10/27/2023, 7:06 PM

Hey All, trying to write unit tests for pyflink. I can't seem to figure out how to pass the correct envirnoment args for the test framework to find jars and I get this error:

def wrapped_call(self, *args, **kwargs):

> raise TypeError(

"Could not found the Java class '%s'. The Java dependencies could be specified via "

"command line argument '--jarfile' or the config option 'pipeline.jars'" % self._fqn)

E TypeError: Could not found the Java class 'org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration.Builder'. The Java dependencies could be specified via command line argument '--jarfile' or the config option 'pipeline.jars'

I get this when running 'pytest'. Here is what my test looks like:

Copy code

rom pyflink.table import DataTypes, expressions as expr
from pyflink.table.udf import ScalarFunction, udf
from pyflink.testing import source_sink_utils
from pyflink.testing.test_case_utils import PyFlinkStreamTableTestCase, \
    PyFlinkBatchTableTestCase


class PyFlinkStreamUserDefinedFunctionTests(PyFlinkStreamTableTestCase):

    def test_flink(self):
        assert True

So this seems to be an initialization problem. I've tried exporting FLINK_HOME, FLINK_LIB_DIR, FLINK_OPT. Seems like I should be able to do this easily but i'm not finding it in the flink sources either.

David Bryson

10/27/2023, 7:34 PM

I have even tried setting up the jarfile paths manually in my test:

Copy code

class PyFlinkStreamCustomTests(PyFlinkStreamTableTestCase):

    @classmethod
    def setUpClass(cls):
        super(PyFlinkStreamCustomTests, cls).setUpClass()

        CURRENT_DIR=os.path.dirname(os.path.realpath(__file__))
        jar_files = [
            "flink-python-1.16.2-tests.jar",
            "flink-python-1.16.2.jar",
            "log4j-1.2-api-2.17.1.jar",
            "log4j-api-2.17.1.jar",
            "log4j-core-2.17.1.jar",
            "log4j-slf4j-impl-2.17.1.jar"
        ]

        jar_paths = tuple(
            [f"file://{os.path.join(CURRENT_DIR, 'jars', name)}" for name in jar_files]
        )
        <http://logging.info|logging.info>(f"adding local jars - {'; '.join(jar_files)}")
        env.get_config().set("pipeline.jars", ";".join(jar_paths))

    def test_vector_median(self):
        assert True

David Bryson

10/27/2023, 7:34 PM

which gets me a bit farther but the loader cannot find slf4j:

Copy code

E               RuntimeError: Java gateway process exited before sending its port number.
E               Stderr:
E               Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/LoggerFactory
E                at org.apache.flink.client.python.PythonEnvUtils.<clinit>(PythonEnvUtils.java:77)
E                at org.apache.flink.client.python.PythonGatewayServer.main(PythonGatewayServer.java:46)
E               Caused by: java.lang.ClassNotFoundException: org.slf4j.LoggerFactory

David Bryson

10/27/2023, 7:34 PM

pretty weird, especially since I am explicitly passing it

Siddhesh Kalgaonkar

10/27/2023, 7:42 PM

There is no native test framework in Pyflink yet. No documentation either. Its really a pain. You can use

Python Unittest

library to write the unittest cases . Thats what I did.

David Bryson

10/27/2023, 8:12 PM

Well there is these builtin tests which look promising: https://github.com/apache/flink/blob/f8172cdbbc27344896d961be4b0b9cdbf000b5cd/flink-python/pyflink/testing/test_case_utils.py#L172

Siddhesh Kalgaonkar

10/28/2023, 5:13 AM

Well these built-in tests looks promising but doesn’t work well. You will experience it.

David Bryson

10/28/2023, 5:59 AM

Yes I felt it. I threw in the towel and just did the most basic things and it seems to work now.

David Bryson

10/28/2023, 5:59 AM

Copy code

# Initialize the TableEnvironment
        settings = EnvironmentSettings.in_streaming_mode()
        cls.t_env = TableEnvironment.create(settings)

    def test_transformation_with_element_source(self):
        # Create an in-memory table with sample data
        elements = [(1, "foo"), (2, "bar"), (3, "baz")]
        source_table = self.t_env.from_elements(elements, ["id", "name"])

        # Register UDF
        self.t_env.create_temporary_system_function("my_udf", my_udf)

        # Perform transformations
        result_table = source_table.select(col("id"), my_udf(col("name")).alias("name_uppercase"))

        # Collect results for verification
        results = result_table.execute().collect()

This actually worked really well

Siddhesh Kalgaonkar

10/28/2023, 9:29 AM

I didn’t use Table API. I was working with data stream APIs and it’s very tough to test the connectors especially but yeah your answer motivated me to try it once again with the data stream API

🙌 1

David Bryson

10/30/2023, 12:26 AM

One thing that isn't covered anywhere is that it all "just works" if you put flink-conf.yaml in your build directory. Then I can run pytest:

Copy code

pytest -s tests

Siddhesh Kalgaonkar

10/30/2023, 6:52 AM

Cool

David Bryson

10/30/2023, 6:19 PM

One thing i haven't figured out, is how to close things down gracefully. I just get this upon the end of my tests:

[2]  12598 segmentation fault pytest tests/test_example.py

Siddhesh Kalgaonkar

10/31/2023, 6:35 AM

If you are using

PyFlinkStreamTableTestCase

then something must be mentioned in its implementation.

David Bryson

10/31/2023, 3:57 PM

good idea

2 Views

Open in Slack

Previous Next