07.将表转换成DataStream

Leefs 2022-02-25 PM 993℃ 0条

### 一、概念

表可以转换为 DataStream 或 DataSet。这样，自定义流处理或批处理程序就可以继续在Table API 或 SQL 查询的结果上运行了。

将表转换为 DataStream 或 DataSet 时，需要指定生成的数据类型，即要将表的每一行转换成的数据类型。通常，最方便的转换类型就是Row。当然，因为结果的所有字段类型都是明确的，我们也经常会用元组类型来表示。

表作为流式查询的结果，是动态更新的。所以，将这种动态查询转换成的数据流，同样需要对表的更新操作进行编码，进而有不同的转换模式。

### 二、模式介绍

**Table API 中表到 DataStream 有两种模式：**

+ **追加模式（Append Mode）**：用于表只会被插入（Insert）操作更改的场景。

+ **撤回模式（Retract Mode）**：用于任何场景。有些类似于更新模式中 Retract 模式，它只有 Insert 和 Delete 两类操作。

得到的数据会增加一个 Boolean 类型的标识位（返回的第一个字段），用它来表示到底是新增的数据（Insert），还是被删除的数据（老数据， Delete）。

**代码实现如下：**

```java
val resultStream: DataStream[Row] = tableEnv.toAppendStream[Row](resultTable)

val aggResultStream: DataStream[(Boolean, (String, Long))] =
tableEnv.toRetractStream[(String, Long)](aggResultTable)

resultStream.print("result")
aggResultStream.print("aggResult")
```

所以，没有经过 groupby 之类聚合操作，可以直接用 toAppendStream 来转换；而如果经过了聚合，有更新操作，一般就必须用 toRetractDstream。

### 三、Query的解释和执行

Table API 提供了一种机制来解释（Explain）计算表的逻辑和优化查询计划。这是通过 `TableEnvironment.explain（table）`方法或`TableEnvironment.explain（）`方法完成的。

**explain 方法会返回一个字符串，描述三个计划：**

+ 未优化的逻辑查询计划
+ 优化后的逻辑查询计划
+ 实际执行计划

我们可以在代码中查看执行计划：

```java
val explaination: String = tableEnv.explain(resultTable)

println(explaination)
```

Query 的解释和执行过程，老 `planner` 和 `blink planner` 大体是一致的，又有所不同。整 体来讲，Query 都会表示成一个逻辑查询计划，然后分两步解释：

1. 优化查询计划
2. 解释成 DataStream 或者 DataSet 程序

而 Blink 版本是批流统一的，所以所有的 Query，只会被解释成 DataStream 程序；另外在批处理环境 TableEnvironment 下，Blink 版本要到 tableEnv.execute()执行调用才开始解释。

*附参考文章来源：*

*《尚硅谷大数据之Flink》*

标签: Flink

非特殊说明，本博所有文章均为博主原创。

上一篇 06.Table API输出表

下一篇 08.Table API和Flink SQL动态表和持续查询