Handling Deprecated Flink API: Converting Legacy TypeInformation to DataTypes

By 大数据技术与架构 – Scenario: After Flink 1.9 the Type API was deprecated and replaced by DataTypes, but the legacy Type methods that provide schema TypeInformation no longer have a corresponding API.

Problem: Users cannot obtain the required TypeInformation for existing Type definitions, leading to failures when registering TableSinks.

Root cause: The registration method for a TableSink that relied on the old Type API was removed.

Solution: Convert the new DataTypes to legacy TypeInformation using TypeConversions.fromDataTypeToLegacyInfo (or LegacyTypeInfoDataTypeConverter) and then register the TableSink with the converted types.

Example code:

TableSink csvSink = new CsvTableSink("/path/to/file", "|");

String[] fieldNames = {"product", "amount"};

TypeInformation[] fieldTypes = {Types.STRING, Types.INT};

DataType[] newFieldType = {DataTypes.STRING(), DataTypes.INT()};

TypeInformation<?>[] newtypeInformations = TypeConversions.fromDataTypeToLegacyInfo(newFieldType);

// alternative using LegacyTypeInfoDataTypeConverter

// TypeInformation<?> str = LegacyTypeInfoDataTypeConverter.toLegacyTypeInfo(DataTypes.STRING());

// TypeInformation<?> integ = LegacyTypeInfoDataTypeConverter.toLegacyTypeInfo(DataTypes.INT());

// TypeInformation[] newTypInfo = {str, integ};

tableEnv.registerTableSink("RubberOrders", fieldNames, newtypeInformations, csvSink);

The article also includes several screenshots illustrating the issue and the steps taken.

Finally, the author invites readers to share their experiences with Flink upgrades.