FlinkSql之TableAPI详解( 二 ) _生活百科

二、TableAPI读取文件使用TableAPI读取文件时，我们首先需要知道去哪里读取也就是文件路径、读取文件的格式、读取出来的数据的结构也就是结果表的表结构及表名
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(); // } ? } }三、TableAPI 读取、写入Kakfa基本流程
1>需要创建表的运行环境
//创建表的运行环境 StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);2>创建查询出的数据写出结构
//创建表结构 Schema schema=new Schema() .field("id",DataTypes.STRING()) .field("ts",DataTypes.BIGINT()) .field("vc",DataTypes.INT());

FlinkSql之TableAPI详解( 二 )

经验总结扩展阅读

鸿蒙侧边栏怎么删除应用?

淘宝聊天窗口打开失败怎么回事

连衣裙炎热的夏天，穿一件短款修身连衣裙游逛商厦绝对是最惬意的事情！

男人的情感软肋在哪里：3个男人告诉你

桃胶什么季节吃最好

supreme羊驼真假怎么辨别?

28岁女白领：靠出轨38岁领导走向事业顶峰，我却过得很煎熬

久久说情感凤凰男要求AA制，多年后却向妻子求助，妻子回应：我有钱，但不帮

肺结节10个人9个人有吗真的吗

春秋五霸是谁

观赏鱼饲养用水有哪些要求?

月经期的饮食保健

灰紫色的翡翠手镯怎么样

黄色和金黄色的区别?

提拉紧致按摩手法轻柔处理才能改善问题

事业单位高温补贴多少钱事业单位高温补贴发放是每年都有吗

卡西欧手表哪一款性价比高,卡西欧系列的手表都有哪些好的推荐？

护肤先敷面膜还是先用芦荟胶?

蜜蜡和翡翠怎么保养

三星W999有什么配件