Flink table 获取类型信息答案

【问题标题】：Flink table get type informationFlink table 获取类型信息
【发布时间】：2021-12-09 19:57:24
【问题描述】：

我有一个 flink 表，比如CREATE TABLE source(id int, name string) with (...)，还有一个目标表，比如CREATE TABLE destination(id int, unique_name string) with (...)。 unique_name是使用内部flink流程函数中的业务逻辑计算出来的。

所以我们可以安全地假设源模式将与目标模式完全相同（名称和数据类型）。我使用数据流 API 做了一些低级别的 process 以获取 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]

虽然我可以使用以下代码解决此问题，但我想将其泛化并从目标 Table 获取 RowTypeInformation。有没有办法从flinkTable获取TypeInformation。

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

【问题讨论】：

标签： apache-flink flink-streaming

【解决方案1】：

表类型系统比TypeInformation更丰富。如果你可以使用内部类，你可以使用org.apache.flink.table.runtime.typeutils.ExternalTypeInfo。可以使用 Table API 的 DataType 配置为 TypeInformation。

如果您喜欢使用官方支持的 API。您可以使用TypeInformation 声明输入和输出类型，并在调用StreamTableEnvironment.toDataStream(..., DataType) 时使用DataTypes.of(TypeInformation)

【讨论】：

我没有调用toDataStream，看不清楚，下面是我的代码sn-p，你能帮忙。 paste.org/120818
@timo，我使用了 ROW_NAMED 的 externaltypeinfo，但在类 RowRowConverter toInternal 方法中代码仍然失败。这是列排序的问题。但是我添加了带有字段名称的 TypeInformation，但它仍然为空。 tableEnv.fromDataStream(processedDataStream.map(x => x).returns( Types.ROW_NAMED(destinationColumnName, destinationTypeInformation: _*) ))
错误：原因：org.apache.flink.util.FlinkRuntimeException：从外部 DataStream API 到内部 Table API 数据结构的输入转换期间出错。确保在架构中正确声明了提供的配置转换器的数据类型。受影响的记录