Блог

Скажем, у меня есть таблица flink CREATE TABLE source(id int, name string) with (...) и, скажем, таблица назначения CREATE TABLE destination(id int, unique_name string) with (...) . unique_name вычисляется с использованием бизнес-логики во внутренней функции процесса flink.

Таким образом, мы можем с уверенностью предположить, что исходная схема будет точной (имя и типы данных) такой же, как и схема назначения. Я сделал некоторый низкий уровень process , используя API потока данных, чтобы получить destination поток данных. Она имеет outputType вид GenericType<org.apache.flink.types.Row> . Когда я снова конвертирую обратный destination поток данных в таблицу, я получил приведенную ниже ошибку.

 org.apache.flink.table.api.ValidationException: Column types of query result and sink 
for registered table 'default_catalog.default_database.destination' do not match.
Cause: Different number of columns.

Query schema: [f0: RAW('org.apache.flink.types.Row', '...')]
Sink schema:  [id: INT, name: STRING]
 

Хотя я могу решить эту проблему, используя приведенный ниже код, однако я хочу обобщить это и получить RowTypeInformation из пункта назначения Table . Есть ли какой-либо способ получить TypeInformation от flink Table .

 tableEnv.fromDataStream(destionationDataStream.map(x -> x).returns(Types.ROW(Types.Int, Types.String))
 

Система типов таблиц богаче, чем TypeInformation . Если вы согласны с использованием внутренних классов, вы можете использовать org.apache.flink.table.runtime.typeutils.ExternalTypeInfo . Это TypeInformation то, что можно настроить с помощью Table API DataType .

Если вы хотите использовать официально поддерживаемый API. Вы можете объявлять тип ввода и вывода с TypeInformation помощью и использовать DataTypes.of(TypeInformation) при вызове StreamTableEnvironment.toDataStream(..., DataType)