一、表环境配置
/**
表环境的主要功能:
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");
//聚合统计的表输出不能转换为追加流,每来一条数据需要更新,所以要转换为 retractStream
tableEnv.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,输出结果到不同的topic
tableEnv.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 到 Kafka
env.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(); } }