使用StreamTableEnvironment集成JDBCTableSource遇类型转换异常求助

针对你遇到的两个类型转换相关错误，我来一步步帮你解决，适配Flink 1.9.0的版本特性：

问题根源分析 #

1. 第一个ClassCastException：你的PostgreSQL表中connect_time和disconnect_time字段实际类型应该是TIMESTAMP或TIMESTAMPTZ，而你在TableSchema中用了DataTypes.DATE()映射。JDBC驱动会把PostgreSQL的TIMESTAMP类型返回为java.sql.Timestamp，但Flink的LocalDateSerializer只能处理java.time.LocalDate，因此抛出类型转换异常。

2. 第二个TableException：Flink 1.9.0对TIMESTAMP_WITH_TIME_ZONE的类型支持不够完善，无法直接将其反向转换为对应的TypeInformation，所以报错提示不支持这种转换。

解决方案一：调整Schema映射，使用Flink兼容的TIMESTAMP类型 #

这是最简便的方法，直接修改TableSchema中的日期字段类型为DataTypes.TIMESTAMP()，匹配PostgreSQL返回的java.sql.Timestamp类型：

TableSchema tableSchema = TableSchema.builder()
 .field("id", DataTypes.STRING())
 .field("ip", DataTypes.STRING())
 .field("port", DataTypes.INT())
 .field("alias", DataTypes.STRING())
 .field("connect_time", DataTypes.TIMESTAMP()) // 修改为TIMESTAMP类型
 .field("disconnect_time", DataTypes.TIMESTAMP()) // 修改为TIMESTAMP类型
 .build();

这样Flink会使用TimestampSerializer来处理JDBC返回的java.sql.Timestamp对象，避免类型转换错误。如果你的业务确实需要LocalDate，可以在后续流处理中再将Timestamp转换为LocalDate：

tableEnvironment.toAppendStream(table, Row.class)
    .map(row -> {
        // 取出Timestamp并转换为LocalDate
        Timestamp connectTs = (Timestamp) row.getField(4);
        LocalDate connectDate = connectTs.toLocalDateTime().toLocalDate();
        // 同理处理disconnect_time
        Timestamp disconnectTs = (Timestamp) row.getField(5);
        LocalDate disconnectDate = disconnectTs.toLocalDateTime().toLocalDate();
        // 构造新的Row或者直接处理业务逻辑
        return row;
    })
    .print();

解决方案二：自定义JDBCInputFormat，手动控制类型转换 #

如果必须在数据源层面就将日期转换为LocalDate，可以自定义JDBCInputFormat来处理类型映射：

// 自定义JDBCInputFormat
JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()
    .setDrivername("org.postgresql.Driver")
    .setDBUrl("jdbc:postgresql://192.168.4.69:5432/sessiondb")
    .setUsername("postgres")
    .setPassword("postgres")
    .setQuery("SELECT id, ip, port, alias, connect_time, disconnect_time FROM sessions")
    .setRowTypeInfo(new RowTypeInfo(
        BasicTypeInfo.STRING_TYPE_INFO,
        BasicTypeInfo.STRING_TYPE_INFO,
        BasicTypeInfo.INT_TYPE_INFO,
        BasicTypeInfo.STRING_TYPE_INFO,
        LocalDateTypeInfo.INSTANCE, // 指定LocalDate类型
        LocalDateTypeInfo.INSTANCE
    ))
    .setConverter(new JDBCConverter() {
        @Override
        public Object[] toRow(ResultSet resultSet) throws SQLException {
            return new Object[]{
                resultSet.getString("id"),
                resultSet.getString("ip"),
                resultSet.getInt("port"),
                resultSet.getString("alias"),
                // 手动将Timestamp转换为LocalDate
                resultSet.getTimestamp("connect_time").toLocalDateTime().toLocalDate(),
                resultSet.getTimestamp("disconnect_time").toLocalDateTime().toLocalDate()
            };
        }
    })
    .finish();

// 将自定义InputFormat包装成SourceFunction
DataStream<Row> sessionsStream = streamExecutionEnvironment.createInput(jdbcInputFormat);
// 注册为临时表继续使用Table API
tableEnvironment.registerDataStream("sessions", sessionsStream, 
    "id, ip, port, alias, connect_time, disconnect_time");

Table table = tableEnvironment.sqlQuery("SELECT * FROM sessions");
tableEnvironment.toAppendStream(table, Row.class).print();

这种方式完全由你控制ResultSet到Row的转换逻辑，避免了默认JDBCTableSource的类型映射问题。

注意事项 #

• Flink 1.9.0是比较老的版本，对Java 8日期时间API的支持不如后续版本完善，建议如果条件允许可以升级到更高版本（比如1.12+），会有更完善的JDBC类型映射支持。
• 确认你的PostgreSQL表中connect_time和disconnect_time的实际类型：如果是DATE类型，JDBC驱动会返回java.sql.Date，此时用DataTypes.DATE()映射是没问题的；如果是TIMESTAMP/TIMESTAMPTZ，则需要用上述方法处理。

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