[Flink] flinksql and table programming cases

「 This is my participation 2022 For the first time, the third challenge is 28 God , Check out the activity details ：2022 For the first time, it's a challenge 」

One 、 summary

• Flink SQL & Table Background and principle
• The concept of dynamic table
• Commonly used SQL And built-in functions

Why do you need relational API ？

Flink adopt Table API & SQL Realize batch flow unification .

Table API and SQL At the top , yes Flink Provide advanced API operation .Flink SQL yes Flink Real time calculation is a simplified calculation model , A set of standards designed to lower the threshold for users to use real-time computing SQL Semantic development language .

rely on ：

		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-clients_2.11</artifactId>
			<version>${flink.version}</version>
		</dependency>

		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-table-api-java-bridge_2.11</artifactId>
			<version>1.11.1</version>
		</dependency>

		<!--  need  scala api  Then add  -->
		<!--<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-scala-bridge_2.11</artifactId> <version>1.11.1</version> </dependency>-->

		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
			<version>${flink.version}</version>
		</dependency>

		<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-table-planner-blink_2.11</artifactId>
			<version>1.11.1</version>
		</dependency>
 Copy code 

Two 、 principle

In the offline computing scenario Hive Almost half of the offline data processing . Its bottom pair SQL The parsing of the uses Apache Calcite

Flink The same SQL Parsing 、 Optimization and execution teach Calcite.

Apache Calcite perform SQL The main steps of query are as follows ：

1. take SQL Parse into an unchecked abstract syntax tree （AST,Abstract Syntax Tree）. Abstract syntax trees are language independent forms , This is similar to the first step of cross compilation .
2. Validate： verification AST, Main verification SQL Is the statement legal , The result after verification is RelNode Trees .
3. Optimize： Optimize RelNode Tree and generate a physical execution plan .
4. Execute： Translate the physical execution plan into platform specific execution code , Such as Flink Of DataStream Applications for .

Pictured ：

Whether it's a batch query SQL Or streaming query SQL, Will go through the corresponding converter Parser Transform into a node tree SQLNode tree, Then generate a logical execution plan Logical Plan, The logical execution plan generates a physical execution plan that can be executed after optimization , hand DataSet perhaps DataStream Of API To carry out .

A complete Flink Table & SQL Job Also by Source、Transformation、Sink constitute ：

3、 ... and 、 Dynamic table

With the traditional table SQL Compared with ,Flink Table & SQL When processing stream data, it will always be in dynamic data changes , So there is a concept of dynamic table .

The query of dynamic table is the same as that of static table , however , When querying dynamic tables ,SQL Can do continuous query , Will not terminate .

Take a chestnut ,Flink The program accepts a Kafka Stream as input ,Kafka Is the purchase record of the user ：

First ,Kafka The message will be continuously parsed into a growing dynamic table , We execute on the dynamic table SQL New dynamic tables will be generated as result tables .

stay DataStream On the implementation SQL Inquire about , The process is as follows ：

1. DataStream Convert to dynamic table
2. Define persistent queries on dynamic tables
3. Convert the real-time results of continuous queries into dynamic tables
4. Convert the dynamic table representing the query results into DataStream , So that applications can DataStream API Further transform the query results .

Four 、Flink Table & SQL Operators and built-in functions

A simple example ：

SELECT * FROM Table;
SELECT name,age FROM Table;


SELECT name,age FROM Table where name LIKE '% Xiao Ming %';
SELECT * FROM Table WHERE age = 20;
SELECT name, age
FROM Table
WHERE name IN (SELECT name FROM Table2)


SELECT name,age FROM Table where name LIKE '% Xiao Ming %';
SELECT * FROM Table WHERE age = 20;
SELECT name, age
FROM Table
WHERE name IN (SELECT name FROM Table2)


SELECT *
FROM User LEFT JOIN Product ON User.name = Product.buyer

SELECT *
FROM User RIGHT JOIN Product ON User.name = Product.buyer

SELECT *
FROM User FULL OUTER JOIN Product ON User.name = Product.buyer
 Copy code 

window

According to the different partition of window data , at present Apache Flink There are the following 3 Kind of ：

• Scroll the window , Window data has a fixed size , The data in the window is not superimposed ;

• The sliding window , Window data has a fixed size , And there are generation intervals ;

• Session window , Window data has no fixed size , It is divided according to the parameters passed in by the user , Window data is not superimposed ;

1） Scroll the window

Scrolling windows are characterized by ： There's a fixed size 、 The data in the window does not overlap , As shown in the figure below ：

The syntax is as follows ：

SELECT 
    [gk],
    [TUMBLE_START(timeCol, size)], 
    [TUMBLE_END(timeCol, size)], 
    agg1(col1), 
    ... 
    aggn(colN)
FROM Tab1
GROUP BY [gk], TUMBLE(timeCol, size)

 Copy code 

for example , It is necessary to calculate the daily order quantity of each user ：

SELECT user, TUMBLE_START(timeLine, INTERVAL '1' DAY) as winStart, SUM(amount) 
FROM Orders 
GROUP BY TUMBLE(timeLine, INTERVAL '1' DAY), user;
 Copy code 

among ,TUMBLE_START and TUMBLE_END Represents the start time and end time of the window ,TUMBLE (timeLine, INTERVAL '1' DAY) Medium timeLine Represents the column where the time field is located ,INTERVAL '1' DAY Indicates that the time interval is one day .

2） The sliding window

Sliding windows have a fixed size , Unlike scrolling windows, sliding windows can be accessed through slide Parameter controls how often sliding windows are created . It should be noted that , Data overlap may occur in multiple sliding windows , The specific semantics are as follows ：

The syntax of a sliding window is compared with that of a scrolling window , Only one more slide Parameters ：

SELECT 
    [gk], 
    [HOP_START(timeCol, slide, size)] ,
    [HOP_END(timeCol, slide, size)],
    agg1(col1), 
    ... 
    aggN(colN) 
FROM Tab1
GROUP BY [gk], HOP(timeCol, slide, size)

 Copy code 

for example , We have to calculate the past every hour 24 The sales volume of each product in an hour ：

SELECT product, SUM(amount) 
FROM Orders 
GROUP BY HOP(rowtime, INTERVAL '1' HOUR, INTERVAL '1' DAY), product
 Copy code 

In the above case INTERVAL '1' HOUR Represents the time interval of sliding window generation .

3） Session window

The session window defines an inactive time , If no event or message occurs within the specified time interval , The session window closes .

The syntax of the session window is as follows ：

SELECT 
    [gk], 
    SESSION_START(timeCol, gap) AS winStart,
    SESSION_END(timeCol, gap) AS winEnd,
    agg1(col1),
     ... 
    aggn(colN)
FROM Tab1
GROUP BY [gk], SESSION(timeCol, gap)
 Copy code 

give an example , We need to calculate the past of each user 1 Order volume in hours ：

SELECT user, SESSION_START(rowtime, INTERVAL '1' HOUR) AS sStart, SESSION_ROWTIME(rowtime, INTERVAL '1' HOUR) AS sEnd, SUM(amount) 
FROM Orders 
GROUP BY SESSION(rowtime, INTERVAL '1' HOUR), user
 Copy code