Flink 类型机制及 Stream API和Table API类型推断和转换

今天药忘吃喽~ 2024-03-03 08:33 20阅读 0赞

注：本文使用flink 版本是0.13

### 一、类型体系 ###

Flink 有两大API （1）stream API 和 （2）Table API ,分别对应TypeInformation 和 DataType类型体系。

#### 1.1 TypeInformation系统 ####

TypeInformation系统是使用Stream一定会用到的。TypeInformation 以下简称TypeInfo。  
TypeInfo 本质就是一对一的类型映射。在java中一个typeInfo就对应着一个确定的java类型。所以在stream api 中某些情况下。给定数据flink可以根据数据自动推断出TypeInfo。  
但现在Table API大行其道，Flink已经有意在用DataType替代TypeInfo了。所以Flink中有 `DataType To TypeInfo` 的API（`org.apache.flink.table.runtime.typeutils.InternalTypeInfo`虽然只能将DataType转为Table API所规定的TypeInfo），却没有提供 `TypeInfo To DataType` 的API。  
![在这里插入图片描述][e1a7113959f8459c8fc5a110e01b9dd1.png]  
在Stream API中默认使用的承载行数据的类型是`org.apache.flink.types.Row`  
注意：创建流后如果是复杂类型，比Row类型，非标准的Pojo类型，必须明确告诉Flink是审美类型，Flink无法自动推断出的。  
（1）可以在创建流时候提供如`org.apache.flink.streaming.api.environment.StreamExecutionEnvironment#addSource(org.apache.flink.streaming.api.functions.source.SourceFunction<OUT>, java.lang.String, org.apache.flink.api.common.typeinfo.TypeInformation<OUT>)`  
(2) 可以在使用function时候提供，如`org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator#returns(org.apache.flink.api.common.typeinfo.TypeInformation<T>)` 。  
其中return方法参数也有三种（非常重要）：

// 比如 DataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5); 提供 Integer.class就可以了。当然不提供FLink也会推断出的。
    public SingleOutputStreamOperator<T> returns(Class<T> typeClass);
    // 用于创安含有泛型的TypeInfo
    // 如：TypeInformation<Tuple2<String, Double>> info = TypeInformation.of(new TypeHint<Tuple2<String, Double>>(){});
    public SingleOutputStreamOperator<T> returns(TypeHint<T> typeHint);
    // TypeInfo的创建方式
    // 1. 基础类型从org.apache.flink.api.common.typeinfo.BasicTypeInfo选择即可。
    // 2. 不含泛型的类型可从org.apache.flink.api.common.typeinfo.TypeInformation#of方法创建。
    // 3. DataStream<Row> 可由 org.apache.flink.api.java.typeutils.RowTypeInfo 创建
    // 4. DataStream<RowData> 只能由 org.apache.flink.table.runtime.typeutils.InternalTypeInfo#of(org.apache.flink.table.types.logical.RowType)等方法创建。注意创建出来字段的Field TypeInfo的也都是Flink Table API预先指定好的。
    public SingleOutputStreamOperator<T> returns(TypeInformation<T> typeInfo) ;

（3）可以实现接口`org.apache.flink.api.java.typeutils.ResultTypeQueryable`提供。

#### 1.2 DataType系统 ####

Table API 中所有的类型都是围绕着DataType构建的。如`org.apache.flink.table.api.Schema` 和 `org.apache.flink.table.catalog.ResolvedSchema`  
前者代表Table api中的Table对象的表结构，后者代表从catalog中获取的表结构。  
DataType本质由两部分组成：

protected final LogicalType logicalType;
        protected final Class<?> conversionClass;

logicalType 即逻辑类型，是和数据库中的类型对应的。  
conversionClass 即运行时java类型，是实际承载数据的类型。

可以说DataType 与物理类型也是一对一的关系，并有`conversionClass`确定。  
举个例子：  
DataType与数据库中日期对应的类型是`DateType` ，它有一个支持`conversionClass` 的列表。分别支持 `Date` , `LocalDate` ,`Integer`。其中 如果使用`Integer`作为实际承载数据的类型，此时存储的值是与1970-01-01的天数差值。

@PublicEvolving
    public final class DateType extends LogicalType {
        
        private static final String FORMAT = "DATE";
        private static final Set<String> NULL_OUTPUT_CONVERSION = conversionSet(new String[]{
        Date.class.getName(), LocalDate.class.getName(), Integer.class.getName()});
        private static final Class<?> DEFAULT_CONVERSION;
     }

如果需要创建DataType，Flink提供的入口是`org.apache.flink.table.api.DataTypes` 类，提供了各种DataType类型的创建方法。需要注意的是创建的类型都使用了默认 `conversionClass` ,其就是`LogicalType` 实现类中指定的`DEFAULT_CONVERSION`。如果需要指定运行时的类型就需要使用`org.apache.flink.table.types.AbstractDataType#bridgedTo`方法。

而在Table API中默认使用的承载行数据类型是`org.apache.flink.table.data.RowData`,是一个接口。常用的实现类是`org.apache.flink.table.data.GenericRowData`。

在Table API中实际承载数据的类型必须使用FLink指定的类型。比如Date类型必须使用`Integer`，而不能使用java的`LocalDate`类型。具体类型可以从方法 `org.apache.flink.table.types.utils.DataTypeUtils#toInternalDataType(org.apache.flink.table.types.DataType)` 得出。

#### 1.3 Stream API 和 Table API相互转换中的类型 ####

转换为核心是 围绕着实际承载数据的类型 即DataType的`conversionClass` 应为  
转换过程均由TableAPI的核心`org.apache.flink.table.api.bridge.java.StreamTableEnvironment`完成。  
刚刚提及了Schema本质就是DataType，并可由DataType创建`org.apache.flink.table.api.Schema.Builder#fromRowDataType` 。以下就将Schema代指为DataType了。

如下各个方法中  
Stream <–相互转化–>Table 中可以指定DataType，也可以不指定DataType(Schema)。

##### 1.3.1 Stream To Table #####

###### 1.3.1.1 不指定DataType ######

Flink会从DataStream的TypeInfo中推断DataType类型。  
比如 TypeInfo是 `org.apache.flink.api.common.typeinfo.BasicTypeInfo#BYTE_TYPE_INFO` 或者`TypeInfomation.of(Integer.class)` 在创建流时候 是`Integer`类型，则Flink会自动推断出需要使用`DataTypes.INT()` 创建的对象并把其实际的承载类`conversionClass`指定为`Integer.class`。

具体可以参考：[flink 类型推断][flink] 的 `data-type-extraction` 章节：

# 注意在scala中不要使用primitives 类型需要使用包装类型。因为原始类型不允许为空。
    If you intend to implement classes in Scala, it is recommended to use boxed types (e.g. java.lang.Integer) instead of Scala’s primitives. Scala’s primitives (e.g. Int or Double) are compiled to JVM primitives (e.g. int/double) and result in NOT NULL semantics
    # 对于没有被列举的类型，是需要额外提供类型的。比如使用@DataTypeHint
    Other JVM bridging classes mentioned in this document require a @DataTypeHint annotation.

下图是官方的java类型推断成为FLink DataType的类型  
![在这里插入图片描述][b51e706649a946309561aa83b84fe00b.png]

###### 1.3.1.2 指定DataType ######

DataType是有LogicalType的，指定了DataType也就指定了LogicalType逻辑类型。  
比如现在Row中有一个字段的TypeInfo还是 `org.apache.flink.api.common.typeinfo.BasicTypeInfo#BYTE_TYPE_INFO` 或者`TypeInfomation.of(Integer.class)` 也就是java中的int 或 Integer。但指定DataType时候使用`DataTypes.DATE().bridgedTo(Integer.class)`。此时就已经告诉Flink这里我虽然给你提供的是Integer，但实际代表的逻辑是Date日期类型数据了。以后就可以使用Table API所有关于Date日期的转换方法了。

##### 1.3.2 Table To Stream #####

Table中都是包含DataType的，可从方法获得，如`DataType dt = tbl.getResolvedSchema().toPhysicalRowDataType();`

###### 1.3.2.1 不指定DataType ######

不指定情况比较简单Flink Table API 每种LogicalType逻辑类型都有默认的java类型。  
如：

<table> 
 <thead> 
  <tr> 
   <th>TableDataType</th> 
   <th>JavaType(conversionClass)</th> 
   <th>代表内容</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>int</td> 
   <td>Integer.class</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>bigint</td> 
   <td>Long.class</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>date</td> 
   <td>Integer</td> 
   <td>与1900-01-01天数差</td> 
  </tr> 
  <tr> 
   <td>time</td> 
   <td>Integer.class</td> 
   <td>当天的毫秒数</td> 
  </tr> 
 </tbody> 
</table>

所以不指定情况下，得到的DataStream中的原Date日期类型的数据一定会转为Integer.class 。并不是java中常用的 LocalDate.class ，也不是 `LogicalType.DEFAULT_CONVERSION`所规定的`DEFAULT_CONVERSION = LocalDate.class;`。

###### 1.3.2.2 指定DataType ######

指定情况就会在DataStream中获得想要的java类型了。Flink会在此过程中给提供类型转换服务。  
如在表中能获取到DataType，如 `DataType dt = tbl.getResolvedSchema().toPhysicalRowDataType();`  
但在`toDataStream或toChangelogStream` 可以提供DataType。  
当两个DataType不一样的时候Flink就会将Table中的java（物理类型）转换为 提供的。

如下流程：  
（1）使用 `DataStream<Integer>`  
（2）指定`DataTypes.DATE().bridgedTo(Integer.class)`创建Table。此时table中的java数据类型仍然是Integer.class。  
（3）使用Table创建Stream，并指定 `DataTypes.DATE().bridgedTo(LocalDate.class)` ，此时得到的流`DataStream<LocalDate>`。  
借此流程就实现了`DataStream<Integer>` 到 `Stream<LocalDate>` 的转化。

public interface StreamTableEnvironment extends TableEnvironment {
        
    
    <T> Table fromDataStream(DataStream<T> dataStream);
    
    <T> Table fromDataStream(DataStream<T> dataStream, Schema schema);
    
    Table fromChangelogStream(DataStream<Row> dataStream);
    
    Table fromChangelogStream(DataStream<Row> dataStream, Schema schema);
    
    DataStream<Row> toDataStream(Table table);
    
    <T> DataStream<T> toDataStream(Table table, AbstractDataType<?> targetDataType);
    DataStream<Row> toChangelogStream(Table table);
    
    DataStream<Row> toChangelogStream(Table table, Schema targetSchema);
    
    }

### 总结 ###

综上，Flink中比较容易从DataType得到 TypeInfo 。而从TypeInfo中转化为想要的逻辑类型，八成额外提供DataType（如Integer转Date）。所以，还是建议直接使用Table API中的DataType更加方便。

参考文章：  
[Flink之数据类型详解][Flink]  
[Flink类型系统][Flink 1]  
[聊聊Java类型擦除、Flink中使用Lambda表达式丢失信息和Flink类型暗示机制][Java_Flink_Lambda_Flink]  
[聊聊Java泛型类型擦除及Flink类型暗示（type hint）机制][Java_Flink_type hint]  
[Flink数据类型&&序列化&&序列化器][Flink 2]

[e1a7113959f8459c8fc5a110e01b9dd1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/d7ba0e150d64470cb2caad46ca995613.png
[flink]: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/types/#data-type-extraction
[b51e706649a946309561aa83b84fe00b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/03/62aae0d37ec747f389eb52548f6f2deb.png
[Flink]: http://t.csdn.cn/qZdej
[Flink 1]: http://t.csdn.cn/wKZsO
[Java_Flink_Lambda_Flink]: http://t.csdn.cn/ZQs5j
[Java_Flink_type hint]: http://t.csdn.cn/kYySe
[Flink 2]: http://t.csdn.cn/43Dsy