Flink单机和集群环境部署教程

目录

下面是 Apache Flink 的单机和集群环境部署教程，包含部署过程中的注意事项和一个使用案例。Apache Flink 是一个开源的流处理框架，用于处理实时数据流。

一、Flink 单机环境部署

1. 环境准备

• 操作系统：Linux（推荐 Ubuntu 20.04 或 CentOS 7）
• Java：Flink 需要 Java 环境，推荐使用 OpenJDK 8 或 11。
• Flink：下载并安装 Flink。

2. 安装 Java

在 Ubuntu 中：

sudoapt update
sudoaptinstall openjdk-11-jdk

在 CentOS 中：

sudo yum install java-11-openjdk

验证 Java 安装：

java-version

3. 下载并安装 Flink

访问 Flink 官网 下载最新版本的 Flink。

wget https://downloads.apache.org/flink/flink-1.16.2/flink-1.16.2-bin-scala_2.12.tgz
tar-xzvf flink-1.16.2-bin-scala_2.12.tgz
mv flink-1.16.2 /usr/local/flink

4. 配置环境变量

编辑

~/.bashrc

文件，添加以下内容：

exportFLINK_HOME=/usr/local/flink
exportPATH=$PATH:$FLINK_HOME/bin
exportJAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64

应用更改：

source ~/.bashrc

5. 配置 Flink

编辑 Flink 的配置文件，位于

$FLINK_HOME/conf

。

配置

flink-conf.yaml

• taskmanager.memory.process.size：任务管理器的内存配置。
• jobmanager.rpc.address：JobManager 的主机地址。

jobmanager.rpc.address: localhost
taskmanager.memory.process.size: 1024m

配置

masters

文件

在单机模式下，默认配置即可，无需修改。

配置

workers

文件

在单机模式下，默认配置即可，无需修改。

6. 启动 Flink

start-cluster.sh

7. 验证 Flink

访问 Flink Dashboard，确保 Flink 正常运行：

• Flink Dashboard: http://localhost:8081

注意事项

• 确保 Java 和 Flink 环境配置正确。
• 检查防火墙设置，确保 Flink 的端口可访问。
• 内存配置：根据实际需求调整 flink-conf.yaml 中的内存配置。

二、Flink 集群环境部署

1. 环境准备

• 多台服务器：至少 3 台（1 个 JobManager，2 个 TaskManager）。
• 网络：确保各节点之间可以互相访问。
• 操作系统：Linux（Ubuntu 或 CentOS）。
• Java：在所有节点上安装 Java。
• Flink：在所有节点上安装 Flink。

2. 配置 SSH 免密码登录

在 JobManager 节点上生成 SSH 密钥：

ssh-keygen -t rsa

将公钥复制到所有 TaskManager 节点：

ssh-copy-id user@taskmanager1
ssh-copy-id user@taskmanager2

3. 安装 Flink

在所有节点上安装 Flink，步骤与单机安装相同。

4. 配置 Flink 集群

配置

flink-conf.yaml

在所有节点上编辑

$FLINK_HOME/conf/flink-conf.yaml

，配置如下内容：

jobmanager.rpc.address: master
taskmanager.numberOfTaskSlots:2parallelism.default:1taskmanager.memory.process.size: 1024m

• jobmanager.rpc.address：设置为 JobManager 的主机名。
• taskmanager.numberOfTaskSlots：每个 TaskManager 的任务槽数量。
• parallelism.default：默认的并行度。

配置

masters

文件

在 JobManager 节点上编辑

$FLINK_HOME/conf/masters

文件，添加 JobManager 的主机名：

master:8081

配置

workers

文件

在 JobManager 节点上编辑

$FLINK_HOME/conf/workers

文件，添加所有 TaskManager 节点的主机名：

taskmanager1
taskmanager2

5. 启动 Flink 集群

在 JobManager 节点上启动 Flink 集群：

start-cluster.sh

6. 验证集群状态

访问 Flink Dashboard，确保所有节点正常运行：

• Flink Dashboard: http://master:8081

注意事项

• 确保 SSH 配置正确，JobManager 节点需要通过 SSH 无密码访问 TaskManager 节点。
• 内存和资源配置：根据实际需求调整 flink-conf.yaml 中的资源配置。
• 时钟同步：使用 ntpd 或 chrony 确保所有节点的时钟同步。
• 网络配置：确保各节点之间的网络连接正常。

三、Flink 使用案例：Word Count

使用 Java 实现 Flink Word Count

1. 编写 Java 程序

importorg.apache.flink.api.java.ExecutionEnvironment;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.api.java.utils.ParameterTool;importorg.apache.flink.core.fs.FileSystem;importjava.util.Arrays;publicclassWordCount{publicstaticvoidmain(String[] args)throwsException{// 获取运行参数finalParameterTool params =ParameterTool.fromArgs(args);// 设置执行环境finalExecutionEnvironment env =ExecutionEnvironment.getExecutionEnvironment();// 从输入文件中读取数据
        env.readTextFile(params.get("input"))// 拆分成单词.flatMap((String value,Collector<Tuple2<String,Integer>> out)->{Arrays.stream(value.split("\\W+")).forEach(word -> out.collect(newTuple2<>(word,1)));}).returns(TupleTypeInfo.getBasicTupleTypeInfo(String.class,Integer.class))// 按单词分组.groupBy(0)// 统计每个单词的出现次数.sum(1)// 将结果保存到输出文件.writeAsCsv(params.get("output"),FileSystem.WriteMode.OVERWRITE);// 执行程序
        env.execute("Word Count Example");}}

2. 编译并打包 Java 程序

确保你已经安装了 Maven，并在

pom.xml

文件中配置了 Flink 依赖：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.16.2</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.16.2</version></dependency>

编译并打包：

mvn clean package

3. 运行 Java 程序

将输入文件上传到 HDFS 或本地文件系统：

hdfs dfs -put localfile.txt /user/flink/input

运行 Flink 应用：

flink run -c com.example.WordCount target/wordcount-1.0-SNAPSHOT.jar --input /user/flink/input --output /user/flink/output

4. 查看结果

hdfs dfs -cat /user/flink/output/part-00000

使用 Python 实现 Flink Word Count

可以使用 Apache Beam 的 Python API 来实现 Flink 版的 Word Count。

1. 安装 Apache Beam

pip install apache-beam

2. 编写 Python 程序

创建

wordcount.py

：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

defrun():# 创建 PipelineOptions
    options = PipelineOptions(
        runner='FlinkRunner',
        flink_master='localhost:8081',
        environment_type='LOOPBACK')# 创建管道with beam.Pipeline(options=options)as p:# 读取输入文件(p
         |'Read'>> beam.io.ReadFromText('input.txt')|'Split'>> beam.FlatMap(lambda x: x.split())|'PairWithOne'>> beam.Map(lambda x:(x,1))|'GroupAndSum'>> beam.CombinePerKey(sum)|'Write'>> beam.io.WriteToText('output.txt'))if __name__ =='__main__':
    run()

3. 运行 Python 程序

python wordcount.py

4. 查看结果

cat output.txt-00000-of-00001

总结

通过以上步骤，我们成功部署了 Flink 单机和集群环境，并实现了一个简单的 Word Count 应用。Flink 提供了强大的流处理和批处理能力，可以在多种场景下处理实时数据。

部署过程中的注意事项

• 确保 Java 和 Flink 环境配置正确。
• 检查网络配置，确保各节点之间的通信正常。
• 合理分配资源，根据集群规模和任务需求调整 Flink 的资源配置。
• 时钟同步：使用 ntpd 或 chrony 确保所有节点的时钟同步。
• 监控集群状态，定期检查节点状态和资源使用情况，及时处理故障节点。

通过合理配置和优化，Flink 可以在多种场景下提供高效的流处理和批处理能力，为大规模数据处理提供强有力的支持。