PyFlink中使用JsonRowSerializationSchema报错的解决方法

错误根源 #

你遇到的ClassCastException核心原因是类型不匹配：
JsonRowSerializationSchema的作用是将Flink的Row对象序列化为JSON格式的字节流，但你的数据流preprocessed_stream输出的是String类型，序列化器尝试将String强制转换为Row时直接失败。

两种解决方案 #

根据你的实际需求，选择对应方案：

方案一：将数据流输出改为Row类型（适合结构化数据转JSON场景） #

如果你的预处理逻辑是生成结构化数据，需要将其序列化为JSON，那么需要把处理结果包装成Row对象，配合JsonRowSerializationSchema使用：

from pyflink.datastream import MapFunction
from pyflink.types import Row
from pyflink.common.typeinfo import Types
from pyflink.datastream.connectors import KafkaSink, KafkaRecordSerializationSchema
from pyflink.common.serialization import JsonRowSerializationSchema

class Preprocessing(MapFunction):
    def map(self, value):
        # 假设这里处理后得到目标字符串
        processed_str = "your_processed_content"
        # 将字符串包装为Row对象
        return Row(processed_str)

# 调整map算子的输出类型为ROW
preprocessed_stream = ds.map(Preprocessing(), output_type=Types.ROW([Types.STRING()]))

# 序列化器保持原有配置
serialization_schema = JsonRowSerializationSchema.Builder() \
    .with_type_info(Types.ROW([Types.STRING()]))\
    .build()

# 注意：原代码遗漏了sink的.build()方法，必须补上
sink = KafkaSink.builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
            .set_topic("preprocessed_data")
            .set_value_serialization_schema(serialization_schema)
            .build()
    ).build()

preprocessed_stream.sink_to(sink)
env.execute()

方案二：直接发送JSON字符串（适合已生成JSON格式字符串的场景） #

如果你的Preprocessing算子输出的已经是合法的JSON格式字符串，那么不需要使用JsonRowSerializationSchema，直接用SimpleStringSchema就能将JSON字符串发送到Kafka，此时Kafka中存储的消息就是标准JSON格式：

from pyflink.common.serialization import SimpleStringSchema

# 替换序列化器为SimpleStringSchema
sink = KafkaSink.builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
            .set_topic("preprocessed_data")
            .set_value_serialization_schema(SimpleStringSchema())
            .build()
    ).build()

preprocessed_stream.sink_to(sink)
env.execute()

内容的提问来源于stack exchange，提问作者Ania