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

推薦閱讀

交通事故索賠需要準備哪些誤工費證據

光遇12月3日復刻旅行先祖兌換物品要多少蠟燭,高手進階

飛機托運酒水規定2022 國內飛機能帶酒嗎？

宋媽后來怎么樣了宋媽后來怎樣了

知名女星李羲兒車禍縫54針！一臉血跡躺倒在地，車子翻轉掉落田溝 ...

2021年正月初六出生的寶寶名字如何取,簡約大氣旺生肖

今日查詢價格3011元，今日查詢價格3011元寶？

關于閃長巖簡述閃長巖

當一個女人不愛你了會有什么表現

造夢西游4手機版龍幣怎么得到

職場女強人的星座女巨蟹女

報考的c1考了科目一可以改成c2嗎報的是c1,考了科一,想換成c2可以不

李廣的生平事跡李廣人物介紹

平凡的世界讀書心得初二優秀作文700字

2006年屬狗的是什么命五行屬什么

vivox21耗電快怎么辦 vivox21耗電快怎么解決

保衛蘿卜深海16攻略保衛蘿卜深海攻略2關

關于跑跑姜餅人簡述跑跑姜餅人

淡豆豉煮水喝有什么作用與功效淡豆豉煮水喝有什么作用

四個木念什么四個木念什么呀