4.2.4 Flink-流处理框架-Table API 与 SQL-表的查询

太过爱你忘了你带给我的痛 2022-11-07 11:53 170阅读 0赞

**目录**

1.写在前面

2.Table API 的调用

3.SQL 查询

4.代码总结

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

# 1.写在前面 #

利用外部系统的连接器 connector，我们可以读写数据，并在环境的 Catalog 中注册表。 接下来就可以对表做查询转换了。Flink 给我们提供了两种查询方式：Table API 和 SQL。

# 2.Table API 的调用 #

Table API 是集成在 Scala 和 Java 语言内的查询 API。与 SQL 不同，Table API 的查询不会 用字符串表示，而是在宿主语言中一步一步调用完成的。Table API 基于代表一张“表”的 Table 类，并提供一整套操作处理的方法 API。这些方 法会返回一个新的 Table 对象，这个对象就表示对输入表应用转换操作的结果。有些关系型 转换操作，可以由多个方法调用组成，构成链式调用结构。例如 table.select(…).filter(…)，其 中 select（…）表示选择表中指定的字段，filter(…)表示筛选条件。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

代码中的实现如下：

Table sensorTable = tableEnv.from("inputTable");
    Table resultTable = senorTable.select("id, temperature").filter("id ='sensor_1'");

# 3.SQL 查询 #

Flink 的 SQL 集成，基于的是 ApacheCalcite，它实现了 SQL 标准。在 Flink 中，用常规字 符串来定义 SQL 查询语句。SQL 查询的结果，是一个新的 Table。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

代码实现如下：

Table resultSqlTable = tableEnv.sqlQuery("select id, temperature from inputTable where id ='sensor_1'");

当然，也可以加上聚合操作，比如我们统计每个 sensor 温度数据出现的个数，做个 count 统计：

Table aggResultTable = sensorTable.groupBy("id").select("id, id.count as count");

SQL 的实现：

Table aggResultSqlTable = tableEnv.sqlQuery("select id, count(id) as cnt from inputTable group by id");

这里 Table API 里指定的字段，前面加了一个单引号’，这是 Table API 中定义的 Expression 类型的写法，可以很方便地表示一个表中的字段。字段可以直接全部用双引号引起来，也可以用半边单引号+字段名的方式。以后的代码 中，一般都用后一种形式。

# 4.代码总结 #

package com.atguigu.apitest.tableapi;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.DataTypes;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.TableEnvironment;
    import org.apache.flink.table.api.java.BatchTableEnvironment;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.descriptors.Csv;
    import org.apache.flink.table.descriptors.FileSystem;
    import org.apache.flink.table.descriptors.OldCsv;
    import org.apache.flink.table.descriptors.Schema;
    import org.apache.flink.types.Row;
    
    public class TableTest2_CommonApi {
        public static void main(String[] args) throws Exception{
            // 1. 创建环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 1.1 基于老版本planner的流处理
            EnvironmentSettings oldStreamSettings = EnvironmentSettings.newInstance()
                    .useOldPlanner()
                    .inStreamingMode()
                    .build();
            StreamTableEnvironment oldStreamTableEnv = StreamTableEnvironment.create(env, oldStreamSettings);
    
            // 1.2 基于老版本planner的批处理
            ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
            BatchTableEnvironment oldBatchTableEnv = BatchTableEnvironment.create(batchEnv);
    
            // 1.3 基于Blink的流处理
            EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner()
                    .inStreamingMode()
                    .build();
            StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);
    
            // 1.4 基于Blink的批处理
            EnvironmentSettings blinkBatchSettings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner()
                    .inBatchMode()
                    .build();
            TableEnvironment blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings);
    
            // 2. 表的创建：连接外部系统，读取数据
            // 2.1 读取文件
            String filePath = "D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt";
            tableEnv.connect( new FileSystem().path(filePath))
                    .withFormat( new Csv())
                    .withSchema( new Schema()
                    .field("id", DataTypes.STRING())
                            .field("timestamp", DataTypes.BIGINT())
                            .field("temp", DataTypes.DOUBLE())
                    )
                    .createTemporaryTable("inputTable");
    
            Table inputTable = tableEnv.from("inputTable");
    //        inputTable.printSchema();
    //        tableEnv.toAppendStream(inputTable, Row.class).print();
    
            // 3. 查询转换
            // 3.1 Table API
            // 简单转换
            Table resultTable = inputTable.select("id, temp")
                    .filter("id === 'sensor_6'");
    
            // 聚合统计
            Table aggTable = inputTable.groupBy("id")
                    .select("id, id.count as count, temp.avg as avgTemp");
    
            // 3.2 SQL
            tableEnv.sqlQuery("select id, temp from inputTable where id = 'senosr_6'");
            Table sqlAggTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp from inputTable group by id");
    
            // 打印输出
            tableEnv.toAppendStream(resultTable, Row.class).print("result");
            tableEnv.toRetractStream(aggTable, Row.class).print("agg");
            tableEnv.toRetractStream(sqlAggTable, Row.class).print("sqlagg");
    
            env.execute();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/ddf8f00b7ae74dddb71ff81c4650da12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/169082a259164b589b25eae0d05d52b3.png