FlinkSql之TableAPI详解 - ToB企服应用市场:ToB评测及商务社交产业平台

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-csv</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-json</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-compress</artifactId> <version>1.21</version> </dependency>

复制代码

/** * 使用TableAPI的基本套路： * 1.创建表的执行环境 * 2.创建表，将流转换为动态表，表的字段名从bean的属性名自动抽取 * 3.对动态表进行查询 * 4.把动态表转换为流 */

复制代码

DataStream<Row> rowDataStream = tableEnvironment.toAppendStream(result, Row.class);

复制代码

DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnvironment.toRetractStream(select, Row.class);

复制代码

package net.cyan.FlinkSql;
import net.cyan.POJO.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.$;
/** * 使用TableAPI的基本套路： * 1.创建表的执行环境 * 2.创建表，将流转换为动态表，表的字段名从bean的属性名自动抽取 * 3.对动态表进行查询 * 4.把动态表转换为流 */
public class Demo1 {
public static void main(String[] args) {
Configuration configuration=new Configuration();
configuration.setInteger("rest.port",3333);
//创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
env.setParallelism(1);
//模拟数据
DataStreamSource WaterSensorSource = env.fromElements(
new WaterSensor("S1", 1000L, 10),
new WaterSensor("S1", 1000L, 10),
new WaterSensor("S2", 2000L, 20),
new WaterSensor("S3", 3000L, 30),
new WaterSensor("S4", 4000L, 40),
new WaterSensor("S5", 5000L, 50)
);
//创建表的执行环境
StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
//创建表，将流转换为动态表，表的字段名从bean的属性名自动抽取
Table table = tableEnvironment.fromDataStream(WaterSensorSource);
//对表进行查询
//1、过时的查询书写
Table result = table
.where("id='S1'")
.select("*");
//2、不过时的书写
Table result1 = table
// .where($("id").isEqual("S1"))
.select($("id"), $("ts"), $("vc"));
//3.聚合函数
Table select = table
.groupBy($("id"))
.aggregate($("vc").sum().as("sum_vc"))
.select($("id"), $("sum_vc"));
//把动态表转换为流，使用到了之前创建的表运行环境
SingleOutputStreamOperator tuple2DataStream = tableEnvironment
.toRetractStream(select, Row.class)
.filter(t -> t.f0)
.map(t -> t.f1);
// DataStream rowDataStream = tableEnvironment.toAppendStream(result, Row.class);
// DataStream rowDataStream1 = tableEnvironment.toAppendStream(result1, Row.class);
// rowDataStream.print();
// rowDataStream1.print();
tuple2DataStream.print();
try {
//启动执行环境
env.execute();
} catch (Exception e) {
e.printStackTrace();
}
}
}

复制代码

package net.cyan.FlinkSql; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.table.descriptors.Csv; import org.apache.flink.table.descriptors.FileSystem; import org.apache.flink.table.descriptors.Schema; import org.apache.flink.table.types.DataType; import static org.apache.flink.table.api.Expressions.$; public class Demo2_readWriteText { public static void main(String[] args) { //创建执行环境 // Configuration configuration = new Configuration(); // configuration.setInteger("rest.port", 3333); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); StreamTableEnvironment talEnv = StreamTableEnvironment.create(env); //创建查询的数据结果封装类型 Schema schema = new Schema() .field("id", DataTypes.STRING()) .field("ts", DataTypes.BIGINT()) .field("vc", DataTypes.INT()); talEnv .connect(new FileSystem().path("input/sensor.txt")) //读取文件路径 .withFormat(new Csv()) //读取文件的数据格式 .withSchema(schema) //读取出来的数据格式 .createTemporaryTable("sensor");//定义结果表名 //进行查询 Table select = talEnv.from("sensor") .where($("id").isEqual("sensor_1")) .select($("id"), $("ts"), $("vc")); //将查询结果写入到新文件中 //同样建立一个动态表连接 talEnv .connect(new FileSystem().path("input/b.txt")) //写入路径 .withFormat(new Csv()) //写入文件的数据格式 .withSchema(schema) //写入的数据格式 .createTemporaryTable("abc");//定义写入表名 //进行写入操作 select.executeInsert("abc"); // try { // //启动执行环境 // env.execute(); // } catch (Exception e) { // e.printStackTrace(); // } } }

复制代码

//创建表的运行环境 StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

复制代码

//创建表结构 Schema schema=new Schema() .field("id",DataTypes.STRING()) .field("ts",DataTypes.BIGINT()) .field("vc",DataTypes.INT());

复制代码

//创建kafka连接 tabEnv.connect( new Kafka() .version("universal")// 版本号 .property("bootstrap.servers","hadoop102:9092")//地址 .property("group.id","cy")//消费者组 .topic("first")//消费主题 ) .withFormat(new Json())//写入的格式 .withSchema(schema) .createTemporaryTable("a");//临时表

复制代码

//创建表 Table select = tabEnv.from("a").select("*");

复制代码

//创建写入主题 tabEnv.connect( new Kafka() .version("universal")// 版本号 .property("bootstrap.servers","hadoop102:9092")//地址 .topic("first1")//消费主题 .sinkPartitionerRoundRobin()//随机分区 ) .withFormat(new Json())//写入的格式 .withSchema(schema) .createTemporaryTable("c");

复制代码

//写入 select.executeInsert("c");

复制代码

package net.cyan.FlinkSql; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.table.descriptors.Json; import org.apache.flink.table.descriptors.Kafka; import org.apache.flink.table.descriptors.Schema; public class Demo5_readWriteKafka { public static void main(String[] args) { //创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //创建表的运行环境 StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env); //创建表结构 Schema schema=new Schema() .field("id",DataTypes.STRING()) .field("ts",DataTypes.BIGINT()) .field("vc",DataTypes.INT()); //创建kafka连接 tabEnv.connect( new Kafka() .version("universal")// 版本号 .property("bootstrap.servers","hadoop102:9092")//地址 .property("group.id","cy")//消费者组 .topic("first")//消费主题 ) .withFormat(new Json())//写入的格式 .withSchema(schema) .createTemporaryTable("a"); //创建表 Table select = tabEnv.from("a").select("*"); //创建写入主题 tabEnv.connect( new Kafka() .version("universal")// 版本号 .property("bootstrap.servers","hadoop102:9092")//地址 .topic("first1")//消费主题 .sinkPartitionerRoundRobin()//随即分区 ) .withFormat(new Json())//写入的格式 .withSchema(schema) .createTemporaryTable("c"); //写入 select.executeInsert("c"); } }

复制代码