Shulou(Shulou.com)06/02 Report--

This article mainly explains how flink sql calculates how to write pv to mysql in real time. The content in the article is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn how to write pv to mysql in real time.

First of all, we will use datagen to generate test data and randomly generate some user 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" +

")"

Define the sink of mysql, where mysql is the sink of a upsert, so you must have a primary key. When mysql builds the table, we specify the date of the day as the primary key. The mysql ddl is as follows

CREATE TABLE `pv` (

`day_ str` varchar (100) NOT NULL

`pv` bigint (10) DEFAULT NULL

PRIMARY KEY (`day_ str`)

)

The ddl in Flink should match the one in mysql, and the primary key should also be specified.

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" +

")"

Let's write a simple query:

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')")

It may be confusing for students who have been doing batch processing before, for streaming, group by will return a RetractStream, which will be converted to datastream, and you will get a Tuple2 object. If the first field of this object is false, the data is to be withdrawn, true indicates that the data is newly added, and the second field is the actual data. Here, we write the result of this real-time update to mysql. In this way, the mysql table will have only one data every day, and the system will constantly update the pv field.

We can also use flink windows to implement similar requirements, define a window with a window cycle of one day, and then customize a trigger, such as once per second, and write the resulting output to a third-party sink.

Thank you for your reading, the above is the content of "how flink sql calculates how to write pv to mysql in real time". After the study of this article, I believe you have a deeper understanding of how flink sql calculates how to write pv into mysql in real time, and the specific use still needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.