1573 积分	0 好友	207 主题

发消息

Apache StreamPark实战：Flink SQL作业开发与运维一站式指南

发表于 13 小时前 | 查看: 3| 回复: 0

概述

你是否曾被 Flink 应用开发的繁琐配置和复杂运维所困扰？Apache StreamPark 的出现，正是为了将流处理应用的门槛降至冰点。它致力于让开发者从框架细节中解放出来，更专注于业务逻辑的创新。

StreamPark 通过规范化的项目配置、函数式编程引导、最佳实践定义以及丰富的即用型连接器，将配置、开发、测试、部署、监控和运维的全流程标准化。它支持 Scala 和 Java 双语言接口，并构建了一个一站式的流处理作业开发管理平台，为作业从构思到上线的全生命周期提供坚实支撑。

环境预备与安装

1. 环境预备

本次实践环境已预先安装以下组件：

JDK 11.0.30
MySQL 5.7.25-1
Flink 1.20.3
Hadoop 3.4.22

2. 安装 StreamPark

首先，下载并解压安装包：

tar -zxvf apache-streampark_2.12-2.1.7-bin.tar.gz
mv apache-streampark_2.12-2.1.7-bin streampark

配置调整：进入 conf 目录，修改 config.yaml 文件。核心步骤是将默认的 H2 数据源切换为你正在使用的 MySQL 或其他数据库。

StreamPark配置文件数据源设置示例

URL 配置参考如下（请根据实际情况调整）：

jdbc:mysql://localhost:3306/streampark?useSSL=false&useUnicode=true&characterEncoding=UTF-8&allowPublicKeyRetrieval=false&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8

接着，修改本地工作空间路径：
StreamPark工作空间配置示例

数据库初始化：

在你的 MySQL 中创建一个名为 streampark 的数据库。
依次执行 script/schema/mysql-schema.sql 和 script/data/mysql-data.sql 脚本以完成表结构和基础数据的初始化。

驱动放置：将你的 MySQL 驱动 Jar 包（例如 mysql-connector-j-8.0.33.jar）放入 streampark/lib 目录下。

启动服务：

bin/startup.sh

StreamPark服务启动成功日志
启动后，可通过 logs/streampark.out 文件查看运行日志。

访问管理平台

在浏览器中访问 http://your-server-ip:10000。
默认登录账号为 admin，密码为 streampark。
Apache StreamPark登录界面

基础配置

成功登录后，需要进行两项关键的基础配置，以便 StreamPark 能够管理你的 Flink 环境。

1. 设置 Flink 版本

在此处添加你已安装的 Flink 目录，StreamPark 会识别其版本。
Flink版本管理界面

2. 设置 Flink 集群

添加一个 Flink 集群，运行模式选择 standalone，并填写其 Web UI 地址（例如 http://localhost:8081）。配置成功后，点击“眼睛”图标可直接跳转至原生的 Flink Web UI 界面。
Flink集群管理界面

创建并运行你的第一个 Flink SQL 作业

现在，让我们创建一个简单的 Flink SQL 作业来体验整个流程。

进入作业开发：在平台中找到作业开发或类似功能入口，点击“添加”按钮。
填写作业配置：
- 作业类型：选择 Flink SQL。
- 运行模式：选择 standalone。
- Flink 版本：选择你上一步配置好的版本。
- Flink 集群：选择你上一步配置好的集群。
- Flink SQL：在编辑框中填入你的 SQL 脚本。

Flink SQL作业配置界面

这里是一个简单的示例脚本，它创建了一个数据生成源表和一个黑阱接收表，并将数据从源表插入到接收表：

drop table if exists streamsourcetable;
drop table if exists streamsinktable;

create table streamsourcetable (
  content STRING
) WITH (
  ‘connector’ = ‘datagen‘
);

create table streamsinktable (
  content STRING
) WITH (
  ‘connector’ = ‘blackhole‘
);

insert into streamsinktable select content from streamsourcetable;