Hello again. We will have to use our own serialization/deserialization, and our Kafka consumer and its serialization have been implemented in Python. Are there examples of customizing of serialization in Python? There are some examples on the Flink website but all are in Java and Scala. Thanks.

Hi @Amir Hossein Sharifzadeh the Python

SerializationSchema and DeserializationSchema

is just a wrapper of the Java ones, see [1] for more details. You could provide custom Java

SerializationSchema

and

DeserializationSchema

implementations and use them in Python. Actually the built-in

SerializationSchema

and

DeserializationSchema

implementations currently available in PyFlink are also wrappers of the Java ones. See [2] for more details. [1] https://github.com/apache/flink/blob/master/flink-python/pyflink/common/serialization.py#L31 [2] https://github.com/apache/flink/blob/master/flink-python/pyflink/datastream/formats/json.py

Thank you @Dian Fu. In my project, I don’t use consumer but get the messages directly and process them once they arrive. In my code I declare the format of messages (i.e. JSON).

env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(stream_execution_environment=env)
    t_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True)
    t_empadbkgd = f"""
            CREATE TABLE bkgd_table(
                raw_id INT,
                raw_data STRING
            ) WITH (
              'connector' = 'kafka',
              'topic' = 'empadraw',
              'properties.bootstrap.servers' = 'kafka:9092',
              'properties.group.id' = 'MY_GRP',
              'scan.startup.mode' = 'latest-offset',
              'format' = 'json'
            )
            """

So, if I have a deserialize function here, how may I pass it into my table to deserialize the message?

I’m a little confused and have the following questions: • You say that

I don't use consumer but get the messages directly and process them once they arrive

, isn’t the table

bkgd_table

defined in the sample code a Kafka consumer which consumes message from Kafka? • The config option `'format' = 'json'`has already declared that the format of the data stored in Kafka is JSON. It will deserialize it for you atomically when consuming from Kafka. I’m confused that where you want your custom

deserialize function

to be used?

From your first question I mean I don’t call KafkaConsumer directly but yes, we consume messages through tables. From your second question, please ignore ‘format’=‘json’. I am wondering how can I use my deserialize function rather than indicating ‘format’=‘json’. Thanks!

It uses Java SPI mechanism to find the format implementation. You can refer to how `json`format [1] as an example. You can define custom format by implementing interface

DeserializationFormatFactory and SerializationFormatFactory

. Also note that the name of the format is defined here [2]. To use it, you need firstly build a fat jar which contains all the dependencies(refer to json [3] as an example) and then use it in PyFlink as following [4]. After placing it into the classpath via [4], it will find it according to the custom format name using Java SPI mechanism. [1] https://github.com/apache/flink/blob/master/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java#L59 [2] https://github.com/apache/flink/blob/master/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java#L134 [3] https://github.com/apache/flink/blob/master/flink-formats/flink-json/pom.xml#L141 [4] https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/python/dependency_management/#jar-dependencies

@Dian Fu thank you very much. You have been very supportive! I follow your instructions and I hope they would help.

Welcome and feel free to ping me if there are any problems~