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

推荐阅读

华阴市算命最准的在哪

微信上看直播可不可以投影到电视上微信上看直播可不可以投影到电视

图木舒克哪里算命准

带瑞字的女孩名字含义含有瑞字的女孩名字

八字算命：1998年出生的人是什么命

上班期间吵架发病算工伤吗？

清江鱼吃什么长大的清江鱼多大的好吃

手机做泥游戏叫什么名字？做水晶泥游戏攻略[多图]

男孩带星字有寓意的名字

魅族18 pro防水吗

【健康知识】大闸蟹公蟹和母蟹的区别大闸蟹公的还是母的好吃!

写自媒体文章从这3个方面入手自媒体文章该如何写？

十大富贵日柱有哪些

环氧树脂水晶胶配方水晶胶去除方法介绍

今年什么时候打春2020 今年什么时候打春2020几点几分

电动牙刷和普通牙刷的区别区别在这里

重庆巴南哪里可以摘樱桃 2021重庆巴南东温泉桃子采摘攻略

2022年西安社保补缴新规是怎样的？社保可以怎么补缴？

石嘴山市惠农区宇泽副食商行_工商信用信息_经营范围期限状态_法人_地址_注册资本_怎么样

维度强度极性程度是什么意思