Flink CDC系列之：理解学习YARN模式

Apache Hadoop YARN 是许多数据处理框架中流行的资源提供者。Flink 服务提交给 YARN 的 ResourceManager，后者在由 YARN NodeManagers 管理的机器上生成容器。Flink 将其 JobManager 和 TaskManager 实例部署到此类容器中。

Flink 可以根据在 JobManager 上运行的作业所需的处理槽数量动态分配和取消分配 TaskManager 资源。

准备

本入门部分假设从版本 2.10.2 开始有一个可运行的 YARN 环境。最方便的方法是使用 Amazon EMR、Google Cloud DataProc 或 Cloudera 等产品等服务来提供 YARN 环境。不建议在本地或集群上手动设置 YARN 环境以完成本入门教程。

• 通过运行 yarn top 确保您的 YARN 集群已准备好接受 Flink 应用程序。它应该不会显示任何错误消息。
• 从下载页面下载最新的 Flink 发行版并解压。
• 重要提示确保已设置 HADOOP_CLASSPATH 环境变量（可以通过运行 echo $HADOOP_CLASSPATH 进行检查）。如果没有，请使用以下命令进行设置。

export HADOOP_CLASSPATH=`hadoop classpath`

会话模式

Flink 可在所有类 UNIX 环境中运行，即 Linux、Mac OS X 和 Cygwin（适用于 Windows）。

可以参考概述检查支持的版本并下载 Flink 的二进制版本，然后提取存档：

tar -xzf flink-*.tgz

应该设置 FLINK_HOME 环境变量，例如：

export FLINK_HOME=/path/flink-*

在 YARN 上启动 Flink 会话

一旦确保已设置 HADOOP_CLASSPATH 环境变量，即可在 YARN 会话上启动 Flink：

# we assume to be in the root directory of 
# the unzipped Flink distribution# export HADOOP_CLASSPATH
export HADOOP_CLASSPATH=`hadoop classpath`# Start YARN session
./bin/yarn-session.sh --detached# Stop YARN session (replace the application id based 
# on the output of the yarn-session.sh command)
echo "stop" | ./bin/yarn-session.sh -id application_XXXXX_XXX

启动 YARN 会话后，现在可以通过命令输出最后几行中打印的 URL 或通过 YARN ResourceManager Web UI 访问 Flink Web UI。

然后，需要向 flink-conf.yaml 添加一些配置：

rest.bind-port: {{REST_PORT}}
rest.address: {{NODE_IP}}
execution.target: yarn-session
yarn.application.id: {{YARN_APPLICATION_ID}}

{{REST_PORT}} 和 {{NODE_IP}} 应替换为 JobManager Web 界面的实际值，{{YARN_APPLICATION_ID}} 应替换为 Flink 的实际 YARN 应用程序 ID。

设置 Flink CDC

从发布页面下载 Flink CDC 的 tar 文件，然后提取存档：

tar -xzf flink-cdc-*.tar.gz

解压后的 flink-cdc 包含四个目录：bin、lib、log 和 conf。

从发布页面下载连接器 jar，并将其移动到 lib 目录。

下载链接仅适用于稳定版本，SNAPSHOT 依赖项需要您根据特定分支自行构建。

提交 Flink CDC Job

下面是同步整个数据库的示例文件mysql-to-doris.yaml：

################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*server-id: 5400-5404server-time-zone: UTCsink:type: dorisfenodes: 127.0.0.1:8030username: rootpassword: ""pipeline:name: Sync MySQL Database to Dorisparallelism: 2

需要根据自己的需求修改配置文件。最后使用Cli将作业提交到Flink Standalone集群。

cd /path/flink-cdc-*
./bin/flink-cdc.sh mysql-to-doris.yaml

提交成功后返回信息如下：

Pipeline has been submitted to cluster.
Job ID: ae30f4580f1918bebf16752d4963dc54
Job Description: Sync MySQL Database to Doris

可以通过 Flink Web UI 找到正在运行的名为 Sync MySQL Database to Doris 的作业。

请注意，目前不支持提交到应用程序模式集群和 per-job 模式集群。