4.2.3 Flink-流处理框架-Table API 与 SQL-流转表+表转流+创建临时视图（Temporary View）

野性酷女 2022-11-07 11:59 8阅读 0赞

**目录**

1.DataStream流转表Table

1.1 基本概念

1.2 实现代码

1.3 数据类型与Table schema 的对应

2.表转流

2.1 基本概念

2.2 Table API 中表到 DataStream 有两种模式

2.3 实现代码

3.创建临时视图（Temporary View）

--------------------

# 1.DataStream流转表Table #

## 1.1 基本概念 ##

Flink 允许我们把 Table 和 DataStream 做转换：我们可以基于一个 DataStream，先流式 地读取数据源，然后 map 成 POJO，再把它转成 Table。Table 的列字段（column fields），就是 POJO 里的字段，这样就不用再麻烦地定义 schema 了。

## 1.2 实现代码 ##

代码中实现非常简单，直接用 tableEnv.fromDataStream()就可以了。默认转换后的 Table schema 和 DataStream 中的字段定义一一对应，也可以单独指定出来。这就允许我们更换字段的顺序、重命名，或者只选取某些字段出来，相当于做了一次 map 操作（或者 Table API 的 select 操作）。

代码具体如下：

DataStream<String> inputStream = env.readTextFile("sensor.txt");
    DataStream<SensorReading> dataStream = inputStream.map( line -> {String[] fields = line.split(",");
        return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } );
    Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp.rowtime as ts, temperature");

## 1.3 数据类型与Table schema 的对应 ##

在上节的例子中，DataStream 中的数据类型，与表的 Schema 之间的对应关系，是按照类中的字段名来对应的（name-based mapping），所以还可以用 as 做重命名。

Table sensorTable = tableEnv.fromDataStream(dataStream, "timestamp as ts, id as myId, temperature");

Flink 的 DataStream 和 DataSet API 支持多种类型。

组合类型，比如元组（内置 Scala 和 Java 元组）、POJO、Scala case 类和 Flink 的 Row 类型等，允许具有多个字段的嵌套数据结构，这些字段可以在 Table 的表达式中访问。其他类型，则被视为原子类型。

# 2.表转流 #

## 2.1 基本概念 ##

表可以转换为 DataStream 或 DataSet。这样，自定义流处理或批处理程序就可以继续在 Table API 或 SQL 查询的结果上运行了。将表转换为 DataStream 或 DataSet 时，需要指定生成的数据类型，即要将表的每一行转换成的数据类型。通常，最方便的转换类型就是 Row。当然，因为结果的所有字段类型都是明确的，我们也经常会用元组类型来表示。

## 2.2 Table API 中表到 DataStream 有两种模式 ##

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

## 2.3 实现代码 ##

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

DataStream<Row> resultStream = tableEnv.toAppendStream(resultTable, Row.class);
    DataStream<Tuple2<Boolean, Row>> aggResultStream = tableEnv.toRetractStream(aggResultTable, Row.class);
    
    resultStream.print("result");
    aggResultStream.print("aggResult");

# 3.创建临时视图（Temporary View） #

创建临时视图的第一种方式，就是**直接从 DataStream 转换而来**。同样，可以直接对应字段转换；也可以在转换的时候，指定相应的字段。代码如下：

tableEnv.createTemporaryView("sensorView", dataStream);
    tableEnv.createTemporaryView("sensorView", dataStream, "id, temperature, timestamp as ts");

另外，当然还可以**基于 Table 创建视图**：

tableEnv.createTemporaryView("sensorView", sensorTable);

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/cc5246f0200c406fae7db95f4f96f26f.png