Flink: 1.13.2

Tengo un tableEnv de tableEnv que lee datos de transmisión de KafkaSource. Desde este tableEnv, filtro mis datos y los transformo nuevamente en un DataStream.

 DataStream<Row> myStreamData = env.fromSource(source, WatermarkStrategy.noWatermarks(),"mySource"); tableEnv.createTemporaryView("myTable", myStreamData); Table mytable = tableEnv.sqlQuery("select source_timestamp, id from mytable"); DataStream<Row> filteredDatastream = tableEnv.toChangelogStream(myTable);

El tipo de source_timestamp es java.sql.Timestamp y el id es String.

Más tarde, accedo a mis datos desde filteredDatastream y me enfrento al problema de que row.getFieldsAs(source_timestamp) devuelve LocalDateTime en lugar de Timestamp.

 System.out.println(filteredDatastream.getTransformation().getOutputType()); /* Out: ( `source_timestamp` TIMESTAMP(9), // Is a TIMESTAMP `trip_id` STRING ) */ filteredDatastream.map( row -> { System.out.println(row.getFieldAs("source_timestamp").getClass()); /* Out: class java.time.LocalDateTime // (not java.sql.Timestamp) */ return row; } );

¿Cómo puedo hacer que row.getFieldAs("source_timestamp") sea java.sql.Timestamp en lugar de java.time.LocalDateTime sin tener que emitir después? Sabiendo que java.sql.Timestamp parece ser manejado por flink data-type-extraction .