flink sql怎么實時計算當天pv寫入mysql

這篇文章主要講解了“flink sql怎么實時計算當天pv寫入mysql”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“flink sql怎么實時計算當天pv寫入mysql”吧！

首先我們還是使用datagen生成測試數據，隨機生成一些用戶id

     String sourceSql = "CREATE TABLE datagen (\n" +
                " userid int,\n" +
                " proctime as PROCTIME()\n" +
                ") WITH (\n" +
                " 'connector' = 'datagen',\n" +
                " 'rows-per-second'='100',\n" +
                " 'fields.userid.kind'='random',\n" +
                " 'fields.userid.min'='1',\n" +
                " 'fields.userid.max'='100'\n" +
                ")";

定義mysql的sink，這里mysql是作為了一個upsert的sink，所以必須要一個主鍵，在mysql建表的時候我們指定了當天的日期作為主鍵，mysql ddl如下

CREATE TABLE `pv` (
  `day_str` varchar(100) NOT NULL,
  `pv` bigint(10) DEFAULT NULL,
  PRIMARY KEY (`day_str`)
)

Flink中的ddl要和mysql中對的上，也要指定主鍵。

    String mysqlsql = "CREATE TABLE pv (\n" +
                "  day_str STRING,\n" +
                "  pv bigINT,\n" +
                "  PRIMARY KEY (day_str) NOT ENFORCED\n" +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'username' = 'root',\n" +
                "   'password' = 'root',\n" +
                "   'url' = 'jdbc:mysql://localhost:3306/test',\n" +
                "   'table-name' = 'pv'\n" +
                ")";

接下來我們寫一個簡單的查詢：

     tEnv.executeSql("insert into pv SELECT DATE_FORMAT(proctime, 'yyyy-MM-dd') as day_str, count(*) \n" +
                "FROM datagen \n" +
                "GROUP BY DATE_FORMAT(proctime, 'yyyy-MM-dd')");

可能對于以前一直做批處理的同學來說會感到疑惑，對于流式處理來說，group by將會返回一個可撤回流(RetractStream)，轉化成datastream，將會得到一個Tuple2<Boolean, T>對象，這個對象第一個字段如果是false表示數據要撤回，true表示數據是我們新添加的，第二個字段是實際的數據。在這里，我們將這個實時更新的結果寫入到了mysql。這樣mysql表，每天就會只有一個數據，系統會不斷地更新pv字段。

類似的需求我們還可以使用flink的窗口來實現，定義一個窗口周期是一天的窗口，然后自定義一個觸發器，比如每秒鐘觸發一次，然后將結果輸出寫入第三方sink。

感謝各位的閱讀，以上就是“flink sql怎么實時計算當天pv寫入mysql”的內容了，經過本文的學習后，相信大家對flink sql怎么實時計算當天pv寫入mysql這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！