Java 要解决此问题，请使用可序列化类型作为 Beam PTransform 的输出，并配置正确的 Beam 编码器。一些 JDBC 驱动程序可能不支持所有 Java 对象类型。使用 Serializable 或兼容类型可以更好地支持 JDBC 驱动程序。例如，以下示例将 Beam PCollection 中的 Person 对象转换为 Map 并使用 JdbcIO.write（）写入 JDBC 表中：

PCollection persons = ...;

// Convert Person objects to Maps PCollection> personMaps = persons.apply( MapElements.into(TypeDescriptors.maps(TypeDescriptors.strings(), TypeDescriptor.of(Object.class))) .via(Person::toMap));

// Write maps to a JDBC table personMaps.apply(JdbcIO.>write() .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration .create("org.postgresql.Driver", "jdbc:postgresql://host:port/database") .withUsername("username") .withPassword("password")) .withStatement("INSERT INTO person (id, name, age) VALUES (?, ?, ?)") .withPreparedStatementSetter((element, statement) -> { statement.setInt(1, element.get("id")); statement.setString(2, element.get("name")); statement.setInt(3, element.get("age")); }));

Python 在 Python SDK 中，JdbcWriteTransform 在创建操作并调用 JdbcWriteFn.write 方法时，在 Beam 中为每个 PCollection 元素使用依赖于 Python pickle 的默认编码。由于未实现使用这种序列化模式的 Java 编码器实现，因此 Beam 会在创建 coder 时抛出上述异常。

此问题的一种解决方法是将 Java 类型映射到序列化的 Python 对象类型。代码示例：

import apache_beam as beam import psycopg2 import cPickle

class PickledPythonCoder(object): def encode(self, value): return cPickle.dumps(value)

def decode(self, value): return cPickle.loads(value)

def is_deterministic(self): return True

def eq(self, other): return isinstance(other, PickledPythonCoder)

coder = PickledPythonCoder()

class Person(object): def init(self, id, name, age): self.id = id self.name = name self.age = age

def asTuple(self): return (self.id, self.name, self.age)

class WriteToDB(beam.DoFn):