java.sql.Timestamp cannot be cast to org.apache.fl

这个异常通常是由于数据类型不匹配引起的。 在Hive中，Timestamp数据类型存储的时间值以UTC格式表示，而在Flink中，TimestampData数据类型存储的时间值以当前系统时间表示。因此，在将Hive中的Timestamp数据类型转换为Flink中的TimestampData数据类型时，可能会出现类型不匹配的问题。 解决方法之一是将Hive中的Timestamp数据类型转换为相应的字符串类型，然后在Flink中使用字符串数据类型。例如：

scala Copy code SELECT CONCAT('1970-01-01',CAST(timestamp_col AS STRING)) AS timestamp_col FROM your_table; 另一种解决方法是使用自定义的数据类型，例如使用Long类型表示Hive中的Timestamp数据类型。例如：

java Copy code CREATE TABLE your_table ( id INT, timestamp_col TIMESTAMP(3) ) WITH ( 'connector.type' = 'filesystem', 'format.type' = 'csv', 'format.field-delimiter' = ',', 'format.line-delimiter' = '
', 'format.ignore-first-line' = 'true' );

CREATE TABLE your_table_with_custom_timestamp_type AS SELECT id, CONCAT('1970-01-01',CAST(timestamp_col AS STRING)) AS timestamp_col FROM your_table; 然后在Flink中使用自定义的数据类型表示时间值：

scala Copy code val env = StreamExecutionEnvironment.getExecutionEnvironment val tableEnv = StreamTableEnvironment.create(env)

val input = env.fromElements("1", "2020-01-01 00:00:00.000") val output = tableEnv.toAppendStream[(Long, String)].addSink(tableEnv.fromElements("id", "timestamp_col")).setParallelism(1)

input.addSink(output)