大纲
- 安装Flink
- 操作系统
- 安装JDK
- 安装Flink
- 修改配置
- 启动Flink
- 测试
- 使用IntelliJ制作任务包
- 新建工程
- Archetype
- 编写测试代码
- 打包
- 测试
- 参考资料
在《0基础学习PyFlink》专题中，我们熟悉了Flink的相关知识以及Python编码方案。这个系列我们将使用相对主流的Java语言，来实践Flink的相关设计。
安装Flink
操作系统
我安装的是Ubuntu 22TLS版。
cat /proc/version
Linux version 5.15.0-112-generic (buildd@lcy02-amd64-051) (gcc (Ubuntu 11.4.0-1ubuntu1~22.04) 11.4.0, GNU ld (GNU Binutils for Ubuntu) 2.38) #122-Ubuntu SMP Thu May 23 07:48:21 UTC 2024
安装JDK
sudo apt install openjdk-11-jdk
通过下面命令查看java版本号
java --version
openjdk 11.0.23 2024-04-16
OpenJDK Runtime Environment (build 11.0.23+9-post-Ubuntu-1ubuntu122.04.1)
OpenJDK 64-Bit Server VM (build 11.0.23+9-post-Ubuntu-1ubuntu122.04.1, mixed mode, sharing)
安装Flink
从https://flink.apache.org/downloads/可以获得最新的地址
wget https://dlcdn.apache.org/flink/flink-1.19.1/flink-1.19.1-bin-scala_2.12.tgz .
tar -xzf flink-*.tgz
修改配置
为了让本地之外的其他机器可以访问，我们需要修改Flink的配置
vim flink-1.19.1/conf/config.yaml
把所有的localhost改成0.0.0.0
vim的指令是
:%s/localhost/0.0.0.0/g
启动Flink
cd flink-1.19.1/
./bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host fangliang.
Starting taskexecutor daemon on host fangliang.
测试
我们通过IP:8081打开后台，可以看到正常显示
然后我们运行一个软件包自带的任务包
./bin/flink run examples/streaming/WordCount.jar
可以看到如下输出
Executing example with default input data.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
Job has been submitted with JobID 6c309e274649b1585a6be46197b16880
Program execution finished
Job with JobID 6c309e274649b1585a6be46197b16880 has finished.
Job Runtime: 414 ms
以及在后台的任务完成列表中看到该任务
使用IntelliJ制作任务包
我们使用Maven方案。
新建工程
JDK选择和Flink运行环境匹配的Java 11。
Archetype
组ID：org.apache.flink
工件ID：flink-quickstart-java
版本：1.19.1
这个版本号就是我们下载的Flink的版本号。
会生成如下结构的工程
编写测试代码
我们在DataStreamJob.java中插入如下代码：
env.fromData(1, 2, 3, 4, 5)
.map(i -> i * i)
.print();
打包
然后我们看到target目录下生成了jar包
测试
在Flink后台，我们上传这个包
然后运行这个包
可以看到这个包正常运行了
参考资料
- https://flink.apache.org/downloads/
