- **流（DataStream）转换成表（Table）**
**调用fromDataStream()方法**
```java
// 读取数据源
SingleOutputStreamOperator<Sensor> sensor = env.fromSource(...)
// 将数据流转换成表
Table sensorTable = tableEnv.fromDataStream(sensor);
```
由于流中的数据本身就是定义好的POJO类型Sensor，所以我们将流转换成表之后，每一行数据就对应着一个Sensor，而表中的列名就对应着Sensor中的属性。
另外，我们还可以在fromDataStream()方法中增加参数，用来指定提取哪些属性作为表中的字段名，并可以任意指定位置：
// 提取Event中的timestamp和url作为表中的列
Table sensorTable = tableEnv.fromDataStream(sensor, $("id"), $("vc"));
也可以通过表达式的as()方法对字段进行重命名：
// 将timestamp字段重命名为ts
Table sensorTable = tableEnv.fromDataStream(sensor, $("id").as("sid"), $("vc"));
**调用createTemporaryView()方法**
```java
tableEnv.createTemporaryView("sensorTable",sensor, $("id"),$("ts"),$("vc"));
```
调用fromDataStream()方法简单直观，可以直接实现DataStream到Table的转换；不过如果我们希望直接在SQL中引用这张表，就还需要调用表环境的createTemporaryView()方法来创建虚拟视图了。
对于这种场景，也有一种更简洁的调用方式。我们可以直接调用createTemporaryView()方法创建虚拟表，传入的两个参数，第一个依然是注册的表名，而第二个可以直接就是DataStream。之后仍旧可以传入多个参数，用来指定表中的字段。

- **表（Table）转换成流（DataStream）**
**调用toDataStream()方法**
```java
tableEnv.toDataStream(table);
```
**调用toChangelogStream()方法**
```java
// 将表转换成更新日志流
tableEnv.toChangelogStream(table);
```
对于这样有更新操作的表，我们不应该直接把它转换成DataStream打印输出，而是记录一下它的“更新日志”（change log）。这样一来，对于表的所有更新操作，就变成了一条更新日志的流。

- **支持的数据类型**
**原子类型**
在Flink中，基础数据类型（Integer、Double、String）和通用数据类型（也就是不可再拆分的数据类型）统一称作“原子类型”。原子类型的DataStream，转换之后就成了只有一列的Table，列字段（field）的数据类型可以由原子类型推断出。另外，还可以在fromDataStream()方法里增加参数，用来重新命名列字段。
```java
DataStream<Long> stream = ...;
// 将数据流转换成动态表，动态表只有一个字段，重命名为myLong
Table table = tableEnv.fromDataStream(stream, $("myLong"));
```
**Tuple类型**
当原子类型不做重命名时，默认的字段名就是“f0”，这其实就是将原子类型看作了一元组Tuple1的处理结果。
Table支持Flink中定义的元组类型Tuple，对应在表中字段名默认就是元组中元素的属性名f0、f1、f2...。所有字段都可以被重新排序，也可以提取其中的一部分字段。字段还可以通过调用表达式的as()方法来进行重命名。
```java
// 将数据流转换成只包含f1字段的表
Table table = tableEnv.fromDataStream(stream, $("f1"));
// 将数据流转换成包含f0和f1字段的表，在表中f0和f1位置交换
Table table = tableEnv.fromDataStream(stream, $("f1"), $("f0"));
// 将f1字段命名为myInt，f0命名为myLong
Table table = tableEnv.fromDataStream(stream, $("f1").as("myInt"), $("f0").as("myLong"));
```
**POJO 类型**
Flink也支持多种数据类型组合成的“复合类型”，最典型的就是简单Java对象（POJO 类型）。
将POJO类型的DataStream转换成Table，如果不指定字段名称，就会直接使用原始 POJO 类型中的字段名称。POJO中的字段同样可以被重新排序、提却和重命名。
```java
DataStream<Event> stream = ...;
Table table = tableEnv.fromDataStream(stream);
Table table = tableEnv.fromDataStream(stream, $("name"));
Table table = tableEnv.fromDataStream(stream, $("name").as("myName"), $("age").as("myAge"));
```
**Row类型**
Flink中还定义了一个在关系型表中更加通用的数据类型——行（Row），它是Table中数据的基本组织形式。
Row类型也是一种复合类型，它的长度固定，而且无法直接推断出每个字段的类型，所以在使用时必须指明具体的类型信息；我们在创建Table时调用的CREATE语句就会将所有的字段名称和类型指定，这在Flink中被称为表的“模式结构”（Schema）。

ITPub博客

Flink - 表和流的转换

最新文章