使用Flink Table & SQL API来构建批量和流式应用(2)：Table API概述

目录#

Table API 简介
环境准备
表的创建与注册
表的查询与转换
常见操作示例
最佳实践
总结
参考资料

1. Table API 简介#

Flink 的 Table API 是一种用于处理结构化数据的领域特定语言（DSL），它允许用户以编程的方式定义数据处理逻辑。与传统的 SQL 不同，Table API 是一种面向对象的 API，它使用 Java 或 Scala 等编程语言来表达数据处理操作。Table API 提供了一系列的操作符，如选择、过滤、聚合等，可以方便地对表进行转换和处理。

Table API 的主要特点包括：

高层次抽象：用户无需关心底层的执行细节，只需要关注数据处理的逻辑。
类型安全：在编译时检查类型错误，减少运行时错误。
可扩展性：可以与 Flink 的其他 API （如 DataStream API 和 DataSet API）集成，实现更复杂的数据处理。

2. 环境准备#

在开始使用 Flink 的 Table API 之前，需要进行一些环境准备工作。首先，确保你已经安装了 Java 8 或更高版本，并且配置了相应的环境变量。然后，下载并安装 Flink 发行版，解压到指定目录。

接下来，创建一个新的 Maven 项目，并在 pom.xml 文件中添加以下依赖：

<dependencies>
    <!-- Flink Table API -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_2.12</artifactId>
        <version>1.13.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.12</artifactId>
        <version>1.13.2</version>
    </dependency>
 
    <!-- Flink Stream Execution Environment -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.13.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.13.2</version>
    </dependency>
</dependencies>

这些依赖将帮助你引入 Flink 的 Table API 和流处理环境。

3. 表的创建与注册#

在 Flink 的 Table API 中，可以通过不同的方式创建表。以下是几种常见的创建表的方法：

从 DataStream 或 DataSet 创建表#

可以将 Flink 的 DataStream 或 DataSet 转换为表。示例代码如下：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class CreateTableFromDataStream {
    public static void main(String[] args) throws Exception {
        // 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 
        // 创建一个简单的 DataStream
        DataStream<String> dataStream = env.fromElements("hello", "world", "flink");
 
        // 将 DataStream 转换为表
        Table table = tableEnv.fromDataStream(dataStream, "word");
 
        // 注册表
        tableEnv.createTemporaryView("myTable", table);
    }
}

从文件或外部数据源创建表#

可以通过连接器从文件或外部数据源（如 Kafka、HBase 等）创建表。以下是一个从 CSV 文件创建表的示例：

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class CreateTableFromFile {
    public static void main(String[] args) {
        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
 
        // 创建表的 DDL
        String ddl = "CREATE TABLE csvTable (" +
                "  id INT," +
                "  name STRING," +
                "  age INT" +
                ") WITH (" +
                "  'connector' = 'filesystem'," +
                "  'path' = 'file:///path/to/csv/file'," +
                "  'format' = 'csv'" +
                ")";
 
        // 执行 DDL 创建表
        tableEnv.executeSql(ddl);
 
        // 查询表
        Table resultTable = tableEnv.sqlQuery("SELECT * FROM csvTable");
    }
}

4. 表的查询与转换#

创建表后，可以使用 Table API 对表进行查询和转换操作。以下是一些常见的操作：

选择操作#

选择表中的特定列：

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class SelectOperation {
    public static void main(String[] args) {
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
 
        // 假设已经注册了一个名为 myTable 的表
        Table resultTable = tableEnv.from("myTable")
                .select("column1, column2");
    }
}

过滤操作#

根据条件过滤表中的数据：

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class FilterOperation {
    public static void main(String[] args) {
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
 
        // 假设已经注册了一个名为 myTable 的表
        Table resultTable = tableEnv.from("myTable")
                .filter("column1 > 10");
    }
}

聚合操作#

对表中的数据进行聚合：

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class AggregationOperation {
    public static void main(String[] args) {
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
 
        // 假设已经注册了一个名为 myTable 的表
        Table resultTable = tableEnv.from("myTable")
                .groupBy("column1")
                .select("column1, column2.sum as sumColumn2");
    }
}

5. 常见操作示例#

连接操作#

连接两个表：

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class JoinOperation {
    public static void main(String[] args) {
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
 
        // 假设已经注册了两个表：table1 和 table2
        Table resultTable = tableEnv.from("table1")
                .join(tableEnv.from("table2"))
                .where("table1.id = table2.id")
                .select("table1.id, table1.name, table2.age");
    }
}

排序操作#

对表中的数据进行排序：

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 
public class SortOperation {
    public static void main(String[] args) {
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
 
        // 假设已经注册了一个名为 myTable 的表
        Table resultTable = tableEnv.from("myTable")
                .orderBy("column1 asc");
    }
}

6. 最佳实践#

合理使用表的注册：在处理复杂的数据处理逻辑时，合理地注册和使用临时表可以提高代码的可读性和可维护性。
注意性能优化：在进行聚合、连接等操作时，要注意数据的分布和分区，避免数据倾斜等问题。
与 SQL 结合使用：Table API 和 SQL 可以相互补充，对于一些复杂的查询，可以使用 SQL 来表达，而对于一些灵活的转换操作，可以使用 Table API。

7. 总结#

Flink 的 Table API 为用户提供了一种方便、高效的方式来处理结构化数据。通过本博客的介绍，我们了解了 Table API 的基本概念、表的创建与注册、查询与转换操作以及常见的操作示例和最佳实践。希望这些内容能够帮助你更好地使用 Flink 的 Table API 来构建批量和流式应用。

目录#