Hello Team, I have started with Flink with Python few days ago. I'm trying to start with the word count example found on this link but wasn't sure how to test it. Any hints would be appreciated

@Vibhavari Bellutagi Hey, do you mean have to execute it? You can run it in one of the following ways: • locally: just run

python word_count.py

• submit to standalone cluster / yarn cluster / K8s: via

flink run

, see https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/cli/#submitting-pyflink-jobs for more details.

Hi @Dian Fu Thanks for the quick help. I was able to actual test locally. I want to add some unittests to it. I was exploring this Github repo - https://github.com/apache/flink/blob/release-1.16/flink-python/pyflink/testing/test_case_utils.py Can you please guide how to test similar test methods ( like process functions )

@Vibhavari Bellutagi Hey, please take a look at if the following example could help: https://github.com/dianfu/pyflink-faq/tree/main/testing

Great. thanks a lot 🙂

Hey, @Vibhavari Bellutagi you need to install maven which is used here to download the required JAR files. Alternatively, you could also download the JAR file manually here [1] to avoid installing maven and put it under the same directory of test_utils.py. In this case, you could remove this line [2]. [1] https://repo1.maven.org/maven2/org/apache/flink/flink-python/1.16.1/flink-python-1.16.1-tests.jar [2] https://github.com/dianfu/pyflink-faq/blob/main/testing/test_utils.py#L52

Ahh got it, will try this. Thank you 🙂

Re the first issue: I mean https://maven.apache.org/install.html. Re the second issue: It seems that the downloaded JAR file wasn’t loaded correctly. Could you share the following information: • The snapshot which shows that the testing python files and the JAR files are located in the same directory. • The content of test_utils.py, especially this line: https://github.com/dianfu/pyflink-faq/blob/main/testing/test_utils.py#L114 (it makes sure the test JAR files downloaded previously are used) You could also try to print the results of

os.path.pathsep.join(test_jars)

https://github.com/dianfu/pyflink-faq/blob/main/testing/test_utils.py#L111 to check if the paths are correct.

Hi @Dian Fu For the first issue - Maven thing Im yet to try on my local. For Second issue - 1. Please find the attached screenshot. 2. I tried below lines -

print("Construct test classpath")
              print(construct_test_classpath())

output: Construct test classpath /IdeaProjects/ProjectName/core/tests/servicest/flink-python-1.16.1-tests.jar Please let me know, what I'm missing in the above code. Thanks for the patience.

@Vibhavari Bellutagi Current it still reports

TypeError: Could not found the Java class 'org.apache.flink.python.util.DataStreamTestCollectSink'. The Java dependencies could be specified via command line argument '--jarfile' or the config option 'pipeline.jars'

?

Yes

I’m a little confused.

org.apache.flink.python.util.DataStreamTestCollectSink

should exist in

flink-python-1.16.1-tests.jar

. You could verify it as following:

jar tf /IdeaProjects/ProjectName/core/tests/servicest/flink-python-1.16.1-tests.jar | grep DataStreamTestCollectSink

Yes, the class is present, I can see when i run above - org/apache/flink/python/util/DataStreamTestCollectSink.class

@Vibhavari Bellutagi Oh, I guess you need to wrap the following code inside

___init___

, otherwise it will be executed before `PyFlinkStreamingTestCase.___init_`(where we need to do some initialization work, e.g. adding flink-python-1.16.1-tests.jar to classpath, etc) is called.

WordCount(StreamExecutionEnvironment.get_execution_environment()).word_count(
    ["ABC is a good boy. ABC works for an XYZ org"])

Hi @Dian Fu , What I understand from your message, I created a parameterised constructor in wordcount class for the env. When I run the program it still didnt work. And also "I guess you need to wrap the following code inside

___init___"

Do you mean to include the code inside the

__init__

inside the constructor of test or core class?

class WordCountTest(PyFlinkStreamingTestCase):
    def __init__(self):
      super().__init__() # this line is giving an error, it is unable to identify the test
      WordCount(StreamExecutionEnvironment.get_execution_environment()).word_count(["ABC is a good boy. ABC works for an XYZ org"])

When you say the following line, do you mean with the env?

WordCount(StreamExecutionEnvironment.get_execution_environment()).word_count(
    ["ABC is a good boy. ABC works for an XYZ org"])

@Vibhavari Bellutagi I mean it could look this:

from pyflink.datastream.stream_execution_environment import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.common import Types


class WordCount(object):
    def __init__(self, env):
        self.env = env
        self.env.set_runtime_mode(RuntimeExecutionMode.BATCH)
        self.env.set_parallelism(1)

    def word_count(self, input_data):
        text = self.env.from_collection(input_data)
        word_count = text.flat_map(lambda x: str(x).lower().split()) \
            .map(lambda i: (i, 1), output_type=Types.TUPLE([Types.STRING(), <http://Types.INT|Types.INT>()])) \
            .key_by(lambda i: i[0]) \
            .reduce(lambda i, j: (i[0], i[1] + j[1]))
        word_count.print()
        self.env.execute("some-name")




########################################test case ##############################

from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from services.word_count import WordCount
from servicest.test_utils import PyFlinkStreamingTestCase, DataStreamTestSinkFunction

class WordCountTest(PyFlinkStreamingTestCase):
    def setUp(self) -> None:
        super(WordCountTest, self).setUp()
        self.test_sink = DataStreamTestSinkFunction()

    def tearDown(self) -> None:
        self.test_sink.clear()

    def assert_equals_sorted(self, expected, actual):
        expected.sort()
        actual.sort()
        self.assertEqual(expected, actual)

    def test_word_count(self):
        env = StreamExecutionEnvironment.get_execution_environment()
        env.set_runtime_mode(RuntimeExecutionMode.BATCH)
        env.set_parallelism(1)

        # Input data
        input_data = ["ABC is a good boy. ABC works for an XYZ org"]

        # Expected output data
        expected_output = [('abc', 2), ('an', 1), ('for', 1), ('good', 1), ('is', 1), ('org', 1), ('works', 1),
                           ('boy.', 1), ('xyz', 1)]

        # Create WordCount object
        wc = WordCount(env)

        # Create collect sink
        sink = WordCountTest()

        # Execute word_count function
        wc.word_count(input_data).add_sink(sink)

        # Get results from collect sink
        results = self.test_sink.get_results()

        # Assert that the actual output matches the expected output
        self.assert_equals_sorted(results, expected_output)


if __name__ == '__main__':
    WordCount(StreamExecutionEnvironment.get_execution_environment()).word_count(
        ["ABC is a good boy. ABC works for an XYZ org"])

Hi @Dian Fu, Im able to resolve the issue, thanks a lot. I can see the output below -

(a,1)
 (an,1)
 (is,1)
 (abc,2)
 (for,1)
 (org,1)
 (xyz,1)
 (boy.,1)
 (good,1)
 (works,1)
 None

But I have this error below -

wc.word_count(input_data).add_sink(sink)
AttributeError: 'NoneType' object has no attribute 'add_sink'

Can you please let me know why we have NoneType?

@Vibhavari Bellutagi You cannot call

wc.word_count(input_data).add_sink(sink)

, it doesn’t return any thing for ``wc.word_count(input_data)``. You should change

word_count.print()

in the method

WordCount.word_count

to

word_count.add_sink(sink)

Hi @Dian Fu Thanks for replying. Sorry for multiple queries. But I didnt understand, what kind of sink should be passed as an argument? Do I need to write another sync method that could be passed here or is it something else. I am a newbie therefore wanted some clarity. 🙂

@Vibhavari Bellutagi The return type of

wc.word_count(input_data)

is void and it doesn’t return anything. You cannot call

wc.word_count(input_data).add_sink(sink)

, it ’s equal to

None.add_sink

.

@Dian Fu Okay cool I got it now but I am still confused that over here word_count.add_sink(sink) what should be passed instead of sink? Any custom class or any other thing? I am not sure what should be that sink.

@Vibhavari Bellutagi It could accept

sink

implementations which could accept input elements and write them out to external storage. • If you mean the

sink

used in the test case, you could refer to https://github.com/dianfu/pyflink-faq/blob/main/testing/test_datastream_api.py#L41. • If you mean the

sink

in production job, you could refer to connector page: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/datastream/overview/ It has given examples in each connector page how to use it in PyFlink.

Ahh, thanks a lot 🙂 got your point.