一、 Flink Table API & SQL简介
1.1 Table API & SQL的背景
Flink虽然已经拥有了强大的DataStream/DataSet API,而且非常的灵活，但是需要熟练使用Eva或Scala的编程Flink编程API编写程序,为了满足流计算和批计算中的各种场景需求，同时降低用户使用门槛，Flink供- -种关系型的API来实现流与批的统一，那么这就是Flink的Table & SQL API。
自2015年开始，阿里巴巴开始调研开源流计算引擎,最终决定基于Flink打造新一代计算引擎，针对Flink存在的不足进行优化和改进，并且在2019年初将最终代码开源，也就是我们熟知的Blink。Blink 在原来的Flink基础_上最显著的一个贡献就是Flink SQL的实现。
1.2 Table API & SQL的特点
Table & SQL API是-种关系型API,用户可以像操作mysql数据库表一样的操作数据， 而不需要写java代码完成Flink Function,更不需要手工的优化java代码调优。另外，SQL 作为一个非程序员可操作的语言，学习成本很低，如果一个系统提供SQL支持，将很容易被用户接受。
●Table API & SQL是关系型声明式的，是处理关系型结构化数据的
●Table API & SQL批流统一 ，支持stream流计算和batch离线计算
●Table API & SQL查询能够被有效的优化，查询可以高效的执行
●Table API & SQL编程比较容易，但是灵活度没有DataStream/DataSet API和底层Low-leve |API强
二、离线计算TableAPI & SQL
2.1 ●BatchSQLEnvironmept (离线批处理Table API)
public class BachWordCountSQL { public static void main(String[] args) throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); BatchTableEnvironment tEnv = BatchTableEnvironment.create(env); DataSet<WordCount> input = env.fromElements( new WordCount("storm", 1L), new WordCount("flink", 1L), new WordCount("hadoop", 1L), new WordCount("flink", 1L), new WordCount("storm", 1L), new WordCount("storm", 1L) ); tEnv.registerDataSet("wordcount",input,"word,counts"); String sql = "select word,sum(counts) as counts from wordcount group by word" + "having sum(counts) >=2 order by counts desc"; Table table = tEnv.sqlQuery(sql); DataSet<WordCount> result = tEnv.toDataSet(table, WordCount.class); result.print(); } }
2.2 ●BatchTableEnvironmept (离线批处理Table API)
public class BachWordCountTable { public static void main(String[] args) throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); BatchTableEnvironment tEnv = BatchTableEnvironment.create(env); DataSet<WordCount> input = env.fromElements( new WordCount("storm", 1L), new WordCount("flink", 1L), new WordCount("hadoop", 1L), new WordCount("flink", 1L), new WordCount("storm", 1L), new WordCount("storm", 1L) ); Table table = tEnv.fromDataSet(input); Table filtered = table.groupBy("word") .select("word,counts.sum as counts") .filter("counts>=2") .orderBy("counts.desc"); DataSet<WordCount> wordCountDataSet = tEnv.toDataSet(filtered, WordCount.class); wordCountDataSet.print(); } }
执行结果：
三、实时计算TableAPI & SQL
3.1 ●StreamSQLEnvironment (实时流处理Table API)
public class StreamSqlWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //1.实时的table的上下文 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // socket 数据源[hadoop spark flink] DataStreamSource<String> lines = env.socketTextStream("192.168.52.200", 8888); SingleOutputStreamOperator<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String line, Collector<String> out) throws Exception { Arrays.stream(line.split(" ")).forEach(out::collect); } }); //2.注册成为表 tableEnv.registerDataStream("t_wordcount",words,"word"); //3.SQL Table table = tableEnv.sqlQuery("SELECT word,COUNT(1) counts FROM t_wordcount GROUP BY word"); //4.结果 DataStream<Tuple2<Boolean, WordCount>> dataStream = tableEnv.toRetractStream(table, WordCount.class); dataStream.print(); env.execute(); } }
运行结果如下：
3.2 ●StreamTableEnvironment (实时流处理Table API)
//2.注册成为表 Table table = tableEnv.fromDataStream(words, "word"); Table table2 = table.groupBy("word").select("word,count(1) as counts"); DataStream<Tuple2<Boolean, Row>> dataStream = tableEnv.toRetractStream(table2, Row.class); dataStream.print(); env.execute();
四、Window窗口和TableAPI & SQL
4.1 Thumb滚动窗口
实现滚动不同窗口内相同用户的金额计算，将窗口的起始结束时间，金额相加。
数据如下：
1000,user01,p1,5
2000,user01,p1,5
2000,user02,p1,3
3000,user01,p1,5
9999,user02,p1,3
19999,user01,p1,5
程序如下：
public class TumblingEventTimeWindowTable { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); DataStreamSource<String> socketDataStream = env.socketTextStream("192.168.52.200", 8888); SingleOutputStreamOperator<Row> rowDataStream = socketDataStream.map(new MapFunction<String, Row>() { @Override public Row map(String line) throws Exception { String[] fields = line.split(","); Long time = Long.parseLong(fields[0]); String uid = fields[1]; String pid = fields[2]; Double money = Double.parseDouble(fields[3]); return Row.of(time, uid, pid, money); } }).returns(Types.ROW(Types.LONG, Types.STRING, Types.STRING, Types.DOUBLE)); SingleOutputStreamOperator<Row> waterMarkRow = rowDataStream.assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor<Row>(Time.seconds(0)) { @Override public long extractTimestamp(Row row) { return (long) row.getField(0); } } ); tableEnv.registerDataStream("t_orders",waterMarkRow,"atime,uid,pid,money,rowtime.rowtime"); Table table = tableEnv.scan("t_orders") .window(Tumble.over("10.seconds").on("rowtime").as("win")) .groupBy("uid,win") .select("uid,win.start,win.end,win.rowtime,money.sum as total"); tableEnv.toAppendStream(table,Row.class).print(); env.execute(); } }
运行结果如下：
五、Kafka数据源—>Table API & SQL
5.1 KafkaToSQL
public class KafkaWordCountToSql { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); tableEnv.connect(new Kafka() .version("universal") .topic("json-input") .startFromEarliest() .property("bootstrap.servers","hadoop1:9092") ).withFormat(new Json().deriveSchema()).withSchema(new Schema() .field("name", TypeInformation.of(String.class)) .field("gender",TypeInformation.of(String.class)) ).inAppendMode().registerTableSource("kafkaSource"); Table select = tableEnv.scan("kafkaSource").groupBy("gender") .select("gender,count(1) as counts"); tableEnv.toRetractStream(select, Row.class).print(); env.execute(); } }
