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

3> 创建kafka连接
//创建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");//临时表4> 进行查询
//创建表 Table select = tabEnv.from("a").select("*");5> 创建写入kafka连接
//创建写入主题 tabEnv.connect( new Kafka() .version("universal")// 版本号 .property("bootstrap.servers","hadoop102:9092")//地址 .topic("first1")//消费主题 .sinkPartitionerRoundRobin()//随机分区 ? ) .withFormat(new Json())//写入的格式 .withSchema(schema) .createTemporaryTable("c");6> 写入
//写入 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"); ? ? } }

FlinkSql之TableAPI详解( 三 )

经验总结扩展阅读

人活着，一定要明白的10条处世之道

虾40-50是什么意思?

鸡胸肉要不要混着猫粮

写了又删的，是说不出口的心里话

健身|为什么奉劝老年人不要去健身房？56岁阿姨坦言：老公差点和我离婚

赵忠祥|78岁赵忠祥被爆突发疾病，老搭档倪萍前去探望，与家属交谈一直眉头紧锁

2023年7月27日可以提车吗 2023年7月27日是提车日子好吗

跨越年龄之恋的老妻少夫，如今妻子85岁，53岁的丈夫生活举步艰辛

王心凌美容轻松变身小美女

痛楚|【爱茉尔】护肤品怎么样

大家觉得360安全卫士怎么样好用吗（360安全卫士好用吗)

中分|程潇终于掀起“铁刘海”，一身白色羽毛露肩短裙亮相，美到不敢认

测字算卦,2020年农历四月十六日是黄历好日子吗？

汽油和柴油的区别,柴油与汽油的区别

2022年10月1日是黄道吉日吗 2022年10月1日国庆节是多少周年

俗话说：“福不双至人生不顺时，记住这三句话，必将有所改变

2020年女孩免费取名

苏珊|“世界第一女胖子”，被高富帅疯狂追求，婚后生下2个儿子

怎样画唇好看教你简单好掌握的方法

亚洲龙后排座椅怎么放倒？