in PyFlink, is there a way to perform gzip-decompression on events coming from a kinesis or kafka source? Both of these require a SerDe Schema to be passed through, and the only one available is SimpleStringSchema which expects UTF-8 encoding. This does not work for other types of binary messages

Hi Matt, currently PyFlink only wraps schemas available in Java. To work around this, you could define your customized one in Java, add as a dependency through

env.add_jar

, then you can create a java schema at python side and wrap it into

DeserializationSchema

, e.g.

DeserializationSchema(get_gateway().jvm.xxx.GzipDeserializationSchema())

.

eek okay, I’m not great with Java. Are there any prebuilt JARs that might already have this included? I suppose them all I need to do is create a maven build that bundles my dependencies (like the kinesis and dynamodb connectors etc)

A simple way to minimize Java code, I guess, is creating an identity

DeserializationSchema

that keep the

byte[]

message as it is, then you can do decompression on those

bytes

at Python side. Although you can get a

byte[]

serializer by

TypeInformationSerializationSchema(BYTE_PRIMITIVE_ARRAY_TYPE_INFO, ec)

, but it’s length-prefix.