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>

1. flink实战开发----flinkSQL入门大全
2. 使用 Apache Flink 开发实时 ETL
3. [Flink]Flink章3　Flink应用开发 --- Redis Sink
4. 开发第一个Flink应用
5. 使用idea开发flink报错
6. ubuntu下使用apache发布Django应用
7. Effective Apache Flink（二）应用场景
8. Flink 应用
9. Flink 应用篇
10. Mac下使用Phonegap（Apache Cordorva）开发iOS应用
更多相关文章...
• Maven Web 应用 - Maven教程
• XML 应用程序 - XML 教程
• PHP开发工具
• Composer 安装与使用