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详解( 三 )

经验总结扩展阅读

抖音怎么使用自己的音乐?

热情穷逼是什么意思?

干墨鱼一般泡水泡多久?

四两四钱男命婚姻详解一生，四两四钱男命能成大事吗

外婆菜是什么菜腌制的

大蒜长出芽还能吃吗,蒜头发芽了还能吃吗？

给初一新生的一句话需要怎么表达期许

爱慕|女人看你顺眼，不是说喜欢你，而是开始出现这些迹象

发型以家人之名演员染发，唐灿变成熟女人，看到月亮：乖乖女黑化了！

空调质保期一般是几年

汉兰达紫金版是什么？

怎么让孩子表达大胆

炸鸡店名字创意

金华聚佳是什么意思?

汽车暖风怎么关

《情书》：爱情中最遥远的距离，是你不知道我爱你

开空调盖被子是什么心理

作者无笔一个男人的懊悔：娶大龄剩女半年，才知哥们的话不是空穴来风

相思难相拥终不得情的网名

室温25度熟肉放一夜会坏吗