PyFlink使用map函数后将DataStream写入JDBC时报错的解决方法

问题原因 #

报错java.lang.ClassCastException: class [B cannot be cast to class org.apache.flink.types.Row的核心原因是：map操作后的数据类型未被Flink正确识别，默认被序列化为字节数组（[B代表字节数组类型），而JDBC Sink期望接收的是Row类型，导致类型转换失败。即使尝试返回tuple，如果没有显式指定输出类型，Flink仍然无法正确解析为JDBC Sink所需的结构。

解决方案 #

需要做两处关键修改：

• 为map操作显式指定输出类型，确保Flink能识别map后的输出是符合要求的Row类型，和源数据的类型定义保持一致。
• 避免直接修改原Row对象，PyFlink中的Row是不可变结构，直接修改会导致类型异常，应创建新的Row或tuple返回。

修改后完整代码 #

from pyflink.datastream.connectors.jdbc import JdbcSink, JdbcExecutionOptions, JdbcConnectionOptions
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common import Row

JDBC_JAR_PATH = "file:///Users/jar_files/postgresql-42.5.0.jar"

env = StreamExecutionEnvironment.get_execution_environment()

env.add_jars(JDBC_JAR_PATH)
env.set_parallelism(1)

type_info = Types.ROW([Types.INT(), Types.STRING(), Types.STRING(), Types.INT()])
ds = env.from_collection(
    [(101, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),
     (102, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),
     (103, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
     (104, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017)
     ], type_info=type_info).name('Source')


def change_id(data):
    # 创建新的Row对象，避免修改原不可变Row
    return Row.of(data[0] + 100, data[1], data[2], data[3])
    # 也可以返回tuple，效果一致：
    # return (data[0] + 100, data[1], data[2], data[3])


# 关键：为map指定output_type，确保Flink识别输出类型为Row
ds1 = ds.map(change_id, output_type=type_info)

ds2 = ds1.add_sink(
    JdbcSink.sink(
        "insert into books(id, title, authors, year) values (?, ?, ?, ?)",
        type_info,
        JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .with_url('jdbc:postgresql://localhost:5432/nhan_su')
            .with_driver_name('org.postgresql.Driver')
            .with_user_name('psql')
            .with_password('psql')
            .build(),
        JdbcExecutionOptions.builder()
            .with_batch_interval_ms(1000)
            .with_batch_size(200)
            .with_max_retries(5)
            .build()
    ))
env.execute()

关键修改说明 #

• 显式指定map的output_type：通过output_type=type_info参数，告诉Flink map操作后的输出结构和源数据一致，确保数据以Row类型传递给JDBC Sink，而不是被序列化为字节数组。
• 创建新Row返回：使用Row.of()构造新的Row对象，替代直接修改原Row的操作，避免因不可变对象修改导致的类型异常。如果习惯用tuple，返回tuple也是可行的，只要指定了正确的output_type，Flink会自动转换为Row类型。

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