详解 Flink Table API 和 Flink SQL 之表操作

一、表环境配置

/**表环境的主要功能：1.注册 Catalog 和表2.执行 SQL 查询3.注册用户自定义函数（UDF）4.DataStream 和表之间的转换
*/
public class TestTableEnvironment {public static void main(String[] args) throws Exception {//1.配置老版本 planner 的流式处理表环境StreamTableEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);EnvironmentSettings oldStreamSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();StreamTableEnvironment oldStreamTableEnv = StreamTableEnvironment.create(env, oldStreamSettings);//2.配置老版本 planner 的批处理表环境ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();BatchTableEnvironment oldBatchTableEnv = BatchTableEnvironment.create(batchEnv);//3.配置 blink planner 的流式处理表环境EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);//4.配置 blink planner 的批处理表环境EnvironmentSettings blinkBatchSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();TableEnvironment blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings);}
}

二、创建表

1. 表的概念

• Flink 中的表概念是指由多个“行”数据构成的，每个行（Row）可以有定义好的多个列（Column）字段的固定类型的数据组成的二维矩阵
• 表环境可以注册目录 Catalog，并可以基于 Catalog 注册表，会维护一个 Catalog-Table 表之间的 map。表在表环境中有一个唯一的 ID 标识符，由目录（catalog）名，数据库（database）名以及表名三部分组成。在默认情况下，目录名为 default_catalog，数据库名为 default_database，因此直接创建一个表 MyTable 则它的 ID 为 default_catalog.default_database.MyTable
• 表可以是常规的（Table，表），或者虚拟的（View，视图） 。常规表（Table）一般可以用来描述外部数据，比如文件、数据库表或消息队列的数据，也可以直接从 DataStream 转换而来。视图可以从现有的表中创建，通常是 table API 或者 SQL 查询的一个结果

2. 创建方式

2.1 连接器创建

2.1.1 连接文件系统

public class CreateTableFromFileSystem {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//从文件系统读取数据创建表String filePath = "./sensor.txt";tableEnv.connect(new FileSystem().path(filePath)) //ConnectorDescriptor实现类//.withFormat(new OldCsv()) //FormatDescriptor实现类，OldCsv为老版本CSV文件格式.withFormat(new Csv()) //新版本CSV格式需要引入flink-csv依赖.withSchema(new Schema() //传入Schema类实例.field("id", DataTypes.STRING()) //指定表的字段名及类型.field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("sensorTable"); //创建临时表Table table = tableEnv.from("sensorTable");table.printSchema();tableEnv.toAppendStream(table, Row.class).print();env.execute();}
}

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>1.10.1</version>
</dependency>

2.1.2 连接 Kafka

public class CreateTableFromKafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//从Kafka读取数据创建表//Flink的kafka连接器flink-kafka-connector中，1.10版本的已经提供了TableAPI的支持tableEnv.connect(new Kafka() //Kafka 的 ConnectorDescriptor实现类.version("0.11") // 定义 kafka 的版本.topic("sensor") // 定义主题 .property("zookeeper.connect", "localhost:2181").property("bootstrap.servers", "localhost:9092")) .withFormat(new Csv()) //新版本CSV格式需要引入flink-csv依赖.withSchema(new Schema() //传入Schema类实例.field("id", DataTypes.STRING()) //指定表的字段名及类型.field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("kafkaSensorTable"); //创建临时表Table table = tableEnv.from("kafkaSensorTable");table.printSchema();tableEnv.toAppendStream(table, Row.class).print();env.execute();}
}

2.2 虚拟表创建

视图

public class CreateView {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//1.基于 DataStream 创建临时视图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]));});tableEnv.createTemporaryView("sensorView", dataStream);//tableEnv.createTemporaryView("sensorView", dataStream, "id, timestamp as ts, temperature");//2.基于 Table 创建临时视图Table table = tableEnv.fromDataStream(dataStream);tableEnv.createTemporaryView("sensorView2", table);env.execute();}
}

三、查询表

public class TestQueryTable {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//创建表环境StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//注册表tableEnv.connect(new FileSystem().path("./sensor.txt")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("sensor");//获取表Table sensorTable = tableEnv.from("sensor");//查询表//1. 使用 Table API//1.1 简单查询Table resultTable = sensorTable.select("id, temperature").filter("id = 'sensor_6'");//1.2 聚合查询统计Table aggTable = sensorTable.groupBy("id").select("id, id.count as cnt, temperature.avg as avgTemp");//2. 使用 SQL//2.1 简单查询Table sqlResultTable = tableEnv.sqlQuery("select id, temperature from sensor where id = 'sensor_6'");//2.2 聚合统计Table sqlAggTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temperature) as avgTemp from sensor group by id");//打印输出结果tableEnv.toAppendStream(resultTable, Row.class).print("result");tableEnv.toAppendStream(sqlResultTable, Row.class).print("sqlResult");//聚合统计的表输出不能转换为追加流，每来一条数据需要更新，所以要转换为 retractStreamtableEnv.toRetractStream(aggTable, Row.class).print("agg");tableEnv.toRetractStream(sqlAggTable, Row.class).print("sqlAgg");//每一条数据输出两个结果，通过 false 和 true 标识是更新前和更新后的数据env.execute();}
}

四、更新模式

对于流式查询（Streaming Queries），需要声明如何在（动态）表和外部连接器之间执行转换。与外部系统交换的消息类型，由更新模式（update mode）指定

1. 追加模式

Append Mode

• 在追加模式下，表（动态表）和外部连接器只交换插入（Insert）消息

2. 撤回模式

Retract Mode

• 在撤回模式下，表和外部连接器交换的是：添加（Add）和撤回（Retract）消息
• 插入（Insert）会被编码为添加（Add）消息
• 删除（Delete）则编码为撤回（Retract）消息
• 更新（Update）则会编码为，已更新行（上一行）的撤回（Retract）消息，和更新行（新行）的添加（Add）消息
• 在此模式下，不能定义 key

3. 更新插入模式

Upsert Mode

• 在 Upsert 模式下，动态表和外部连接器交换 Upsert 和 Delete 消息
• 这个模式需要一个唯一的 key，通过这个 key 可以传递更新消息。为了正确应用消息，外部连接器需要知道这个唯一 key 的属性
• 插入（Insert）和更新（Update）都被编码为 Upsert 消息
• 删除（Delete）编码为 Delete 信息
• 这种模式和 Retract 模式的主要区别在于，Update 操作是用单个消息编码的，所以效率会更高

五、输出表

表的输出是通过将数据写入到 TableSink 来实现的。TableSink 是 Table API 中提供的一个向外部系统写入数据的通用接口，可以支持不同的文件格式（比如 CSV、Parquet）、存储数据库（比如 JDBC、HBase、Elasticsearch）和消息队列（比如 Kafka）

1. 输出到文件系统

public class TestInsertIntoTable {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//创建表环境StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//注册表tableEnv.connect(new FileSystem().path("./sensor.txt")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("sensor");//获取表Table sensorTable = tableEnv.from("sensor");//查询表//1.1 简单查询Table resultTable = sensorTable.select("id, timestamp, temperature").filter("id = 'sensor_6'");//1.2 聚合查询统计Table aggTable = sensorTable.groupBy("id").select("id, id.count as cnt, temperature.avg as avgTemp");//注册输出表tableEnv.connect(new FileSystem().path("./out.txt")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("cnt", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("out");//输出数据resultTable.insertInto("out"); //使用CSVTableSink，实现AppendStreamTableSink接口//aggTable.insertInto("out"); //聚合统计因为不是追加操作，不能直接输出到文件系统env.execute();}
}

2. 输出到 Kafka

public class TestKafkaTableOutput {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//1.连接 Kafka 读取数据tableEnv.connect(new Kafka().version("0.11").topic("sensor").property("zookeeper.connect", "localhost:2181").property("bootstrap.servers", "localhost:9092")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("inputTable");//2.查询转换Table sensorTable = tableEnv.from("inputTable");//1.1 简单查询Table resultTable = sensorTable.select("id, temperature").filter("id = 'sensor_6'");//1.2 聚合查询统计Table aggTable = sensorTable.groupBy("id").select("id, id.count as cnt, temperature.avg as avgTemp");//3.连接Kafka，输出结果到不同的topictableEnv.connect(new Kafka().version("0.11").topic("sinkTest").property("zookeeper.connect", "localhost:2181").property("bootstrap.servers", "localhost:9092")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("outputTable");resultTable.insertInto("outputTable");//使用KafkaTableSinkBase，实现AppendStreamTableSink 接口//注意：聚合统计的表也不能直接 insertInto 到 Kafkaenv.execute();}
}

3. 输出到 ES

• 引入 flink-json 依赖

  <dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>1.10.1</version>
  </dependency>
  

• 实现

  /**1. ElasticSearch 的 connector 可以在 upsert 模式下操作，可以使用 Query 定义的键（key）与外部系统交换 UPSERT/DELETE 消息。2. 对于“仅追加”（append-only）的查询，ElasticSearch 的 connector 可以在 append 模式下操作，这样就可以与外部系统只交换 insert 消息。3. es 目前支持的数据格式，只有 Json
  */
  public class TestESTableOutput {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//注册表tableEnv.connect(new FileSystem().path("./sensor.txt")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("sensor");//获取表Table sensorTable = tableEnv.from("sensor");//聚合查询统计Table aggTable = sensorTable.groupBy("id").select("id, id.count as cnt, temperature.avg as avgTemp");//建立与 ES 的连接tableEnv.connect(new Elasticsearch().version("6").host("localhost", 9200, "http").index("sensor").documentType("temp")).inUpsertMode().withFormat(new Json()).withSchema(new Schema().field("id", DataTypes.STRING()).field("cnt", DataTypes.BIGINT()).field("avgTemp", DataTypes.DOUBLE())).createTemporaryTable("esOutputTable");aggTable.insertInto("esOutputTable");env.execute();}
  }
  

4. 输出到 MySQL

• 引入 Table API 的 jdbc 连接依赖

  <dependency><groupId>org.apache.flink</groupId><artifactId>flink-jdbc_2.12</artifactId><version>1.10.1</version>
  </dependency>
  

• 实现

  /**jdbc 连接的代码实现比较特殊，因为没有对应的 java/scala 类实现 ConnectorDescriptor，所以不能直接 tableEnv.connect()。可以使用 Flink SQL 的执行 DDL 的接口 tableEnv.sqlUpdate()
  */
  public class TestMysqlTableOutput {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//注册表tableEnv.connect(new FileSystem().path("./sensor.txt")).withFormat(new Csv()).withSchema(new Schema().field("id", DataTypes.STRING()).field("timestamp", DataTypes.BIGINT()).field("temperature", DataTypes.DOUBLE())).createTemporaryTable("sensor");//获取表Table sensorTable = tableEnv.from("sensor");//聚合查询统计Table aggTable = sensorTable.groupBy("id").select("id, id.count as cnt, temperature.avg as avgTemp");//建立与 Mysql 的连接String sinkDDL= "create table jdbcOutputTable (" +" id varchar(20) not null, " +" cnt bigint not null, " +" avgTemp double not null " +") with (" +" 'connector.type' = 'jdbc', " + " 'connector.url' = 'jdbc:mysql://localhost:3306/test', "  +" 'connector.table' = 'sensor_count', " +" 'connector.driver' = 'com.mysql.jdbc.Driver', " +" 'connector.username' = 'root', " +" 'connector.password' = '123456' )";tableEnv.sqlUpdate(sinkDDL);  //执行 DDL 创建表aggTable.insertInto("jdbcOutputTable");env.execute();}
  }