5、Flink SQL管理平台flink-streaming-platform-web安装搭建

一、下载和环境

1、下载源码

无情（朱慧培）/flink-streaming-platform-web

2、环境要求

• hadoop版本 2+
• flink 版本 1.12.0 官方地址:

Gitee.com

• jdk版本 jdk1.8
• mysql版本 5.6+
• scala版本 2.12

scala安装

• kafka版本 1.0+

Kafka安装

二、编译

目前web客户端支持的flink版本是1.15.3,如果需要调整flink版本可下载源码 然后修改pom里面的版本号

vim flink-streaming-platform-web-master\pom.xml
1

<flink.version>1.17.0</flink.version>
<scala.binary.version>2.12</scala.binary.version>
1
2

可能调整后导致flink引用的上下不兼容 需要你手动解决下
保存后打包

mvn clean package  -Dmaven.test.skip=true
1

错误:::danger
在windows上执行assembly任务，但是它包含了相对于linux的路径
:::

修改flink-streaming-platform-web-master\deployer\src\main\assembly\dev.xml

最后打好的包在 /flink-streaming-platform-web-master/deployer/target
包名是：flink-streaming-platform-web.tar.gz

三、部署

1、将打包好的文件放入到172服务器上，由于依赖Flink，必须与Flink同一服务器

2、解压

cd /home
tar -xvzf /home/flink-streaming-platform-web.tar.gz
1
2

3、修改配置文件：application.properties

vim /home/flink-streaming-platform-web/conf/application.properties
1

####jdbc信息
server.port=9084
spring.datasource.url=jdbc:mysql://10.9.70.172:3306/flink_web?serverTimezone=CTT&characterEncoding=utf8&useUnicode=true&useSSL=false&autoReconnect=true&zeroDateTimeBehavior=round&allowMultiQueries=true&allowPublicKeyRetrieval=true
spring.datasource.username=root
spring.datasource.password=Stwc.2wsx

1
2
3
4
5
6
7

4、启动服务

cd /home/flink-streaming-platform-web/bin/
sh deploy.sh  start
1
2

5、访问

9084端口在浏览器访问
http://10.9.70.172:9084/
登录号：admin 密码 123456

用户密码过于简单，可以通过修改数据库中表user的密码
下为按照源码提供的加密方式加密的密码，密码为“Stwc.2wsx”

185c9df7a8eedd40e3ec2a2404941549
1

登录成功

6、配置flink web平台

页面上点击系统设置，进入配置页面：

/opt/software/flink/flink-1.17.0/
/home/flink-streaming-platform-web/
http://10.9.70.166:8088/
http://10.9.70.172:8081/
http://10.9.70.172:8081/;http://10.9.70.166:8081/
1
2
3
4
5

:::warning
这里的参数意义：

• Flink客户端目录：就是安装的Flink目录；
• Flink管理平台目录：就是下载的flink-streaming-platform-web放的目录；
• yarn RM http地址：就是yarn.resourcemanager.webapp.address，通常是8088端口；

经过测试，配置这3个参数即可使用。
:::

7、添加flink插件

Just a moment…

上传到三台服务器 /opt/software/flink/flink-1.17.0/lib/ 上

7、运行demo

在172的mysql数据库flink_web中创建表

CREATE TABLE sync_test_1 (
  `day_time` varchar(64) NOT NULL,
  `total_gmv` bigint(11) DEFAULT NULL,
  PRIMARY KEY (`day_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
1
2
3
4
5

在web页面新建sql流任务：

-p 2 -Dtaskmanager.numberOfTaskSlots=2 -Dyarn.application.queue=root.default
1

create table flink_test_1 ( 
    id BIGINT,
    day_time VARCHAR,
    amnount BIGINT,
    proctime AS PROCTIME ()
)
with ( 
    'connector' = 'kafka',
    'topic' = 'flink-connector',
    'properties.bootstrap.servers' = 'JSYF3:9092', 
    'properties.group.id' = 'flink_gp_test1',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json',
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true'
);
CREATE TABLE sync_test_1 (
    day_time string,
    total_gmv bigint,
    PRIMARY KEY (day_time) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://10.9.70.172:3306/flink_web?characterEncoding=UTF-8',
    'table-name' = 'sync_test_1',
    'username' = 'root',
    'password' = 'Stwc.2wsx'
);
INSERT INTO sync_test_1 
SELECT day_time,SUM(amnount) AS total_gmv
FROM flink_test_1
GROUP BY day_time;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

创建好任务后，启动任务吧。

启动后可在yarn的8088端口页面看到起了一个application，名称是新建任务填写的名称加flink@前缀
错误:::danger
1：Caused by: java.lang.NoClassDefFoundError: org/apache/kafka/clients/consumer/OffsetResetStrategy
:::
这是缺少kafka-clients的jar包，只需将下载的kafka-clients jar包放在lib目录下重启flink，提交任务
:::danger
2：Caused by: org.apache.flink.yarn.YarnClusterDescriptor$YarnDeploymentException: The cluster does not have the requested resources for the JobManager available!
Maximum Memory: 1024MB Requested: 4096MB. Please check the ‘yarn.scheduler.maximum-allocation-mb’ and the ‘yarn.nodemanager.resource.memory-mb’ configuration values
:::
内存不够，修改Flink的配置

vim /opt/software/hadoop/hadoop-3.3.5/etc/hadoop/yarn-site.xml
1

    <property>
        <name>yarn.nodemanager.resource.memory-mb</name>
        <value>4096</value>
    </property>

    <!-- 每个容器请求的最小内存资源（以MB为单位）。-->

    <property>

        <name>yarn.scheduler.minimum-allocation-mb</name>

        <value>512</value>

    </property>

    <!-- 每个容器请求的最大内存资源（以MB为单位）。-->

   <property>

        <name>yarn.scheduler.maximum-allocation-mb</name>

        <value>2048</value>

    </property>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

重启yarn
任务也启动成功

kafka控制台往主题里面写数据，主题不存在会自动创建

/home/kafka/kafka_2.12-3.4.0/bin/kafka-console-producer.sh --broker-list 10.9.70.172:9092 --topic flink-connector
{"day_time":"2021-01-01","amnount":"2022","id":1}
{"day_time":"2021-01-01","amnount":"2022","id":1}
{"day_time":"2021-01-01","amnount":"2022","id":2}
{"day_time":"2021-01-02","amnount":"2000","id":3}
1
2
3
4
5

/home/kafka/kafka_2.12-3.4.0/bin/kafka-run-class kafka.tools.ConsumerOffsetChecker --zookeeper JSYF3:2181 --group flink_gp_test1 --topic flink-connector
cd /home/kafka/kafka_2.12-3.4.0/bin/
./kafka-consumer-groups.sh --describe --group flink_gp_test1 --zookeeper JSYF3:2181

./kafka-topics.sh --list --bootstrap-server JSYF3:9092
1
2
3
4
5

【问题】

1、如何找到Flink错误日志

如果启动失败
查看文件

/opt/software/flink/flink-1.17.0/log/flink-root-client-jsyf3.log
1

如果还是没有具体的错误信息

提示让你如看application中的文件
http://10.9.70.166:8088/cluster/apps

2、Flink yarn Could not start rest endpoint on any port in port range 8081

此错误是说端口被占用。
对应的配置是 flink-conf.yaml中的rest.bind-port。
rest.bind-port不设置，则Rest Server默认绑定到rest.port端口(8081)。
rest.bind-port可以设置成列表格式如50100,50101，也可设置成范围格式如50100-50200。推荐范围格式，避免端口冲突。

vim /opt/software/flink/flink-1.17.0/conf/flink-conf.yaml 
1

rest.bind-port: 8080-8090
1

【参考】
Dinky
部署 | Dinky

Flink SQL管理平台flink-streaming-platform-web安装搭建
Flink SQL管理平台flink-streaming-platform-web安装搭建