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

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详解】

FlinkSql之TableAPI详解( 三 )

推荐阅读

饭圈拉瓜是什么意思饭圈拉郎是什么意思

山药圆子怎么做才好吃，肉末山药圆子的做法？

父母买房可以写未成年孩子的名字吗,规避转让税但不能贷款

不倒翁小鸟的制作教程

“魑魅魍魉” 你们知道怎么读吗？魑魅魍魉怎么读

关于淮北公交简述淮北公交

excel查找不出来是什么原因

强势老人是家庭的灾难【老人强势发生家庭矛盾】

糖尿病人适合送的礼品女总裁给男秘书送粥

大红袍分几种肉桂

熟黄芪泡水喝的功效与作用熟黄芪泡水喝有哪些功效与作用

柠檬片的功能与功效柠檬片的功效作用

防弹咖啡怎么做好喝防弹咖啡怎么做

今年广东调整养老金的新消息 2022广东省养老金上调方案及补发时间

昆明机场离哪个高铁站近，昆明飞机场到高铁站有多远？

孕妇铅过高有什么症状有什么不良影响

毓字取名女孩有寓意属兔最搭的名字有哪些？

关于特别认真地生活简述特别认真地生活

办银行卡需要什么办银行卡需要什么证件和材料

终不似少年游终不似少年游