Apache Flink-使用FlinkSQL開發應用

時間 2020-07-20

標籤 apache flink 使用 flinksql 開發應用欄目 Apache 简体版

原文原文鏈接

這個模塊作的是實時統計用戶每10分鐘內的搜索次數，也就是10分鐘級別的搜索頻率。用戶搜索時，服務端會把搜索數據發送的Kafka中。javascript

直接看Flink的部分吧，這部分作的事情就是消費Kafka中的數據而後基於Event Time（事件時間）的10分鐘級別的滾動窗口統計搜索次數。而後將結果集sink到mysql中。這個功能實現起來仍是比較簡單的，再加上用FlinkSQL來作就更容易了。先看代碼吧css

2 具體實現html

2.1核心代碼部分java

建立源表
CREATE TABLE KAFKA_SOURCE_SEARCH_DATA (
data VARCHAR,
ts timestamp(3),
WATERMARK FOR ts as ts - INTERVAL '5' SECOND
) WITH (
'connector.type' = 'kafka',
'connector.version' = 'universal',
'connector.properties.group.id' = 'group-flink',
'connector.topic' = 'search_data',
'connector.startup-mode' = 'earliest-offset',
'connector.properties.zookeeper.connect' = 'localhost:2181',
'connector.properties.bootstrap.servers' = 'kafka1:9094',
'format.type' = 'json'
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
建立目標表
CREATE TABLE MYSQL_SINK_SEARCH_FREQUENCY (
cnt BIGINT,
cnt_time timestamp(3)
) WITH (
'connector.type' = 'jdbc',
'connector.url' = 'jdbc:mysql://123.207.11.229:3306/search_lite?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&autoReconnect=true&failOverReadOnly=false',
'connector.table' = 'user_search_frequency',
'connector.username' = 'root',
'connector.password' = 'yxyj6900@',
'connector.write.flush.max-rows' = '1'
)
1
2
3
4
5
6
7
8
9
10
11
計算邏輯
INSERT INTO MYSQL_SINK_SEARCH_FREQUENCY(cnt,cnt_time)
SELECT
COUNT(*) as cnt,
TUMBLE_START(ts, INTERVAL '10' MINUTE) as cnt_time
FROM KAFKA_SOURCE_SEARCH_DATA
GROUP BY TUMBLE(ts, INTERVAL '10' MINUTE)
1
2
3
4
5
6
發送到Kafka的JSON樣本
{"data": "8-1受集度爲q的均布載荷做用的矩形槓桿截面簡支梁","ts": "2020-06-06T15:01:39.780Z"}
1
其實思路仍是蠻清晰，先設計好源表和目標表，而後在設計計算邏輯。關於源表和目標表的建立以及FlinkSQL的使用，官網提供了詳細的文檔，這裏就再也不贅述了。mysql

這裏須要提的是關於時間的格式，時間格式要遵循Flink支持的格式，一開始我用的其餘時間格式致使報了莫名其妙的錯誤。ajax

private static final SimpleDateFormat df=new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
1
嗯，其實寫到這裏，基本上就算完成了，能夠直接使用Flink提供的sql-client提交就好了注意須要手動安裝須要的依賴。但我這裏使用的Java開發打包Jar提交的（後面再來寫經過sql-client提交任務）。sql

2.2 Java代碼部分apache

使用官方提供的模板來建立項目，先看須要的依賴（部分代碼）json

<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>1.10.0</version>
<scope>provided</scope>
</dependency>bootstrap

<!--kafka-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-json</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!--mysql-->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.19</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-jdbc_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>