此页面由 Cloud Translation API 翻译。

Dataproc 可选 Flink 组件

您可以在创建 Dataproc 时激活其他组件，例如 Flink 使用可选组件功能。本页面介绍了如何创建 Dataproc 集群使用 Apache Flink 激活可选组件（Flink 集群），然后在该集群上运行 Flink 作业。

您可以使用 Flink 集群执行以下操作：

使用 Dataproc Jobs 资源运行 Flink 作业从 Google Cloud 控制台、Google Cloud CLI 或 Dataproc API 登录。
使用 flink CLI 运行 Flink 作业在 Flink 集群主服务器节点上运行的
在 Flink 上运行 Apache Beam 作业
在 Kerberos 化的集群上运行 Flink

创建 Dataproc Flink 集群

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 Dataproc 创建具有 Flink 组件的 Dataproc 集群的 API 集群上已激活的资源

建议：使用带有 Flink 组件的标准单主实例虚拟机集群。 Dataproc 高可用性模式集群（具有 3 个主虚拟机）不支持 Flink 高可用性模式。

控制台

如需使用 Google Cloud 控制台创建 Dataproc Flink 集群，请执行以下操作：请执行以下步骤：

打开 Dataproc 在 Compute Engine 上创建 Dataproc 集群页面。
1. 已选择设置集群面板。
  1. 在版本控制部分中，确认或更改 Image Type and Version。集群映像版本决定了安装在集群上的 Flink 组件版本。
    - 映像版本必须为 1.5 或更高版本才能在集群上激活 Flink 组件（请参阅支持的 Dataproc 版本来查看每个 Dataproc 映像版本）。
    - 如需运行 Flink 作业，映像版本必须为 [TBD] 或更高版本（请参阅运行 Dataproc Flink 作业）。
  2. 在组件部分中，执行以下操作： <ph type="x-smartling-placeholder">
    1. 在组件网关下，选择 启用组件网关。您必须启用组件网关激活指向 Flink 历史记录服务器界面的组件网关链接。启用组件网关也会启用对 Flink Job Manager 网页界面的访问权限运行在 Flink 集群上的应用。
    2. 在可选组件下，选择 Flink 和其他可选项在集群上激活的组件
2. 点击自定义集群（可选）面板。
  1. 在集群属性部分，点击添加属性 每个可选集群属性以添加到集群您可以添加 flink 前缀属性在 /etc/flink/conf/flink-conf.yaml 中配置 Flink 属性，将用作您在集群上运行的 Flink 应用的默认值。
    
    示例：
    - 设置flink:historyserver.archive.fs.dir 指定 Cloud Storage 位置以写入 Flink 作业历史记录此位置将由运行 Flink History Server 的）。
    - 使用 flink:taskmanager.numberOfTaskSlots=n 设置 Flink 任务槽。
  2. 在自定义集群元数据部分中，点击添加元数据以添加元数据。例如，添加 flink-start-yarn-session true：用于运行 Flink YARN 守护程序 (/usr/bin/flink-yarn-daemon) 在集群主节点的后台运行节点启动 Flink YARN 会话（请参阅 Flink 会话模式）。
3. 如果您使用的是 Dataproc 映像 2.0 或更低版本，点击管理安全性（可选）面板，然后在项目访问权限下，选择Enables the cloud-platform scope for this cluster。创建集群时，cloud-platform 范围默认处于启用状态（使用 Dataproc 映像 2.1 或更高版本）。
点击创建以创建集群。

gcloud

如需使用 gcloud CLI 创建 Dataproc Flink 集群，请运行以下命令 gcloud gclid clusters create 命令行中的命令 Cloud Shell：

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --image-version=DATAPROC_IMAGE_VERSION \
    --optional-components=FLINK \
    --enable-component-gateway \
    --properties=PROPERTIES
    ... other flags

注意：

CLUSTER_NAME：指定集群的名称。
REGION：指定 Compute Engine 区域集群所在的位置
DATAPROC_IMAGE_VERSION：（可选）指定映像版本以便在集群上使用集群映像版本决定了安装在集群上的 Flink 组件版本。
- 映像版本必须为 1.5 或更高版本才能在集群上激活 Flink 组件（请参阅支持的 Dataproc 版本来查看每个 Dataproc 映像版本）。
- 如需运行 Flink 作业，映像版本必须为 [TBD] 或更高版本（请参阅运行 Dataproc Flink 作业）。
--optional-components：您必须指定 FLINK 组件才能运行 Flink 和集群上的 Flink HistoryServer 网络服务。
--enable-component-gateway：您必须启用组件网关激活与 Flink 历史记录服务器界面的组件网关链接。启用组件网关也会启用对 Flink Job Manager 网页界面，运行于 Flink 集群。
PROPERTIES.（可选）指定一个或多个集群属性。
- 使用如下过滤条件创建 Dataproc 集群时：映像版本 2.0.67+ 和 2.1.15+，则可以使用 --properties 标志在 /etc/flink/conf/flink-conf.yaml 中配置 Flink 属性，用作您在集群上运行的 Flink 应用的默认值。
- 您可以将 flink:historyserver.archive.fs.dir 设置为指定 Cloud Storage 位置以写入 Flink 作业历史记录此位置将由运行 Flink History Server 的）。
- 多个属性示例：
```
--properties=flink:historyserver.archive.fs.dir=gs://my-bucket/my-flink-cluster/completed-jobs,flink:taskmanager.numberOfTaskSlots=2
```
其他标志：
- 您可以添加可选的 --metadata flink-start-yarn-session=true 标志，用于运行 Flink YARN 守护程序 (/usr/bin/flink-yarn-daemon) 在集群主服务器节点的后台启动 Flink YARN 会话（请参阅 Flink 会话模式）。
使用 2.0 或更早版本的映像时，您可以将 --scopes=https://www.googleapis.com/auth/cloud-platform 旗帜允许您的集群访问 Google Cloud API （请参阅范围最佳实践）。创建集群时，cloud-platform 范围默认处于启用状态（使用 Dataproc 映像 2.1 或更高版本）。

API

要使用 Dataproc API 创建 Dataproc Flink 集群，请提交 clusters.create 请求，如下所示：

注意：

将 SoftwareConfig.Component 发送至 FLINK。
您可以选择性地设置 SoftwareConfig.imageVersion 指定要在集群上使用的映像版本。集群映像版本决定了安装在集群上的 Flink 组件版本。
- 映像版本必须为 1.5 或更高版本才能在集群上激活 Flink 组件（请参阅支持的 Dataproc 版本来查看每个 Dataproc 映像版本）。
- 如需运行 Flink 作业，映像版本必须为 [TBD] 或更高版本（请参阅运行 Dataproc Flink 作业）。
设置 EndpointConfig.enableHttpPortAccess 更改为 true，以启用组件网关指向 Flink 历史记录服务器界面的链接。启用组件网关也会启用对 Flink Job Manager 网页界面，运行于 Flink 集群。
您可以选择性地设置 SoftwareConfig.properties 指定一个或多个集群属性。
- 您可以指定充当角色的 Flink 属性默认配置。例如：您可以设置 flink:historyserver.archive.fs.dir 指定 Cloud Storage 位置以写入 Flink 作业历史记录此位置将由运行 Flink History Server 的）。
您可以选择设置：
- GceClusterConfig.metadata. 例如，指定 flink-start-yarn-session true 以运行 Flink YARN 守护程序 (/usr/bin/flink-yarn-daemon) 在集群主服务器实例的后台运行节点启动 Flink YARN 会话（请参阅 Flink 会话模式）。
- GceClusterConfig.serviceAccountScopes 到 https://www.googleapis.com/auth/cloud-platform（cloud-platform 个范围）启用对 Google Cloud 的访问权限时（使用 2.0 或更早的映像版本）由集群提供的 API（请参阅范围最佳实践）。创建集群时，cloud-platform 范围默认处于启用状态（使用 Dataproc 映像 2.1 或更高版本）。

创建 Flink 集群后

使用Flink History Server 组件网关查看在 Flink 集群上运行的 Flink 历史记录服务器。
在组件网关中使用 YARN ResourceManager link 查看 Flink Job Manager 网页界面在 Flink 集群上运行。
创建 Dataproc 永久性历史记录服务器查看由现有和已删除的 Flink 集群写入的 Flink 作业历史记录文件。

使用 Dataproc `Jobs` 资源运行 Flink 作业

您可以使用 Google Cloud 控制台中的 Dataproc Jobs 资源来运行 Flink 作业， Google Cloud 控制台、Google Cloud CLI 或 Dataproc API。

控制台

如需从控制台提交示例 Flink WordCount 作业，请执行以下操作：

打开 Dataproc 提交作业页面（位于在浏览器中打开 Google Cloud 控制台。
填写提交作业页面上的字段：
1. 从集群列表中选择集群名称。
2. 将作业类型设置为 Flink。
3. 将主类或 jar 设置为 org.apache.flink.examples.java.wordcount.WordCount。
4. 将 Jar 文件设置为 file:///usr/lib/flink/examples/batch/WordCount.jar。
  - file:/// 表示位于集群上的文件。Dataproc 在创建 Flink 集群时安装了 WordCount.jar。
  - 此字段还接受 Cloud Storage 路径 (gs://BUCKET/JARFILE) 或 Hadoop 分布式文件系统 (HDFS) 路径 (hdfs://PATH_TO_JAR).
点击提交。
- 作业驱动程序输出会显示在作业详情页面上。
- Flink 作业列在 Dataproc 作业页面 Google Cloud 控制台中。
- 从作业或作业详情页面中点击停止或删除停止或删除作业。

gcloud

如需将 Flink 作业提交到 Dataproc Flink 集群，请运行 gcloud CLI gcloud dataproc 作业提交命令行中的命令 Cloud Shell。

gcloud dataproc jobs submit flink \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --class=MAIN_CLASS \
    --jar=JAR_FILE \
    -- JOB_ARGS

注意：

CLUSTER_NAME：指定 Dataproc Flink 的名称将作业提交到哪个集群
REGION：指定 Compute Engine 区域集群所在的位置
MAIN_CLASS：指定您的 API 的 main 类。 Flink 应用，例如： <ph type="x-smartling-placeholder">
- org.apache.flink.examples.java.wordcount.WordCount
JAR_FILE：指定 Flink 应用 jar 文件。您可以指定： <ph type="x-smartling-placeholder">
- 使用 file:///安装在集群上的 jar 文件前缀：
  - file:///usr/lib/flink/examples/streaming/TopSpeedWindowing.jar
  - file:///usr/lib/flink/examples/batch/WordCount.jar
- Cloud Storage 中的 jar 文件： gs://BUCKET/JARFILE
- HDFS 中的 jar 文件： hdfs://PATH_TO_JAR
JOB_ARGS：（可选）在双短划线 (--) 后面添加作业参数。

提交作业后，作业驱动程序输出将显示在或 Cloud Shell 终端

Program execution finished
Job with JobID 829d48df4ebef2817f4000dfba126e0f has finished.
Job Runtime: 13610 ms
...
(after,1)
(and,12)
(arrows,1)
(ay,1)
(be,4)
(bourn,1)
(cast,1)
(coil,1)
(come,1)

REST

本部分介绍如何将 Flink 作业提交到 Dataproc Flink 使用 Dataproc jobs.submit API。

您可以将 clusterLabels 字段添加到如下所示的 API 请求中，以指定一个或多个集群标签。Dataproc 会将作业提交到集群集群标签匹配的所有对象（请参阅 jobs.submit API 了解详情）。

在使用任何请求数据之前，请先进行以下替换：

PROJECT_ID：Google Cloud 项目 ID
REGION：集群地区
CLUSTER_NAME：指定要将作业提交到的 Dataproc Flink 集群的名称

HTTP 方法和网址：

POST https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit

请求 JSON 正文：

{
  "job": {
    "placement": {
      "clusterName": "CLUSTER_NAME"
    },
    "flinkJob": {
      "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
      "jarFileUris": [
        "file:///usr/lib/flink/examples/batch/WordCount.jar"
      ]
    }
  }
}

如需发送您的请求，请展开以下选项之一：

curl（Linux、macOS 或 Cloud Shell）

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI，或者使用了 Cloud Shell，这会使您自动登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit"

PowerShell (Windows)

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit" | Select-Object -Expand Content

您应该收到类似以下内容的 JSON 响应：

</ph>

{
  "reference": {
    "projectId": "PROJECT_ID",
    "jobId": "JOB_ID"
  },
  "placement": {
    "clusterName": "CLUSTER_NAME",
    "clusterUuid": "CLUSTER_UUID"
  },
  "flinkJob": {
    "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
    "args": [
      "1000"
    ],
    "jarFileUris": [
      "file:///usr/lib/flink/examples/batch/WordCount.jar"
    ]
  },
  "status": {
    "state": "PENDING",
    "stateStartTime": "2020-10-07T20:16:21.759Z"
  },
  "jobUuid": "JOB_UUID"
}

Flink 作业列在 Dataproc 作业页面 Google Cloud 控制台中。
您可以从作业或作业详情页面中点击停止或删除。来停止或删除作业。

使用 `flink` CLI 运行 Flink 作业

而不是使用 Dataproc Jobs 资源运行 Flink 作业；您可以使用 flink CLI 在 Flink 集群的主节点上运行 Flink 作业。

以下部分介绍了运行 flink CLI 作业的不同方法 Dataproc Flink 集群。

通过 SSH 连接到主节点：使用 SSH 实用程序打开终端窗口。
设置类路径：从 Flink 集群主服务器虚拟机：
```
export HADOOP_CLASSPATH=$(hadoop classpath)
```
注意： Flink 命令语法可能会因 Flink 而异安装在 Dataproc 集群上的版本。请参阅 Dataproc 映像版本列表或在集群上运行 flink --version 以检查已安装的 Flink 组件版本运行 Flink 集群。如需获取其他标志信息，请运行 flink 命令帮助。
运行 Flink 作业：您可以在不同的环境中运行 Flink 作业 YARN 上的部署模式：应用模式、每个作业模式和会话模式。
1. 应用模式：Dataproc 映像 2.0 及更高版本支持 Flink 应用模式。此模式会在 YARN 作业管理器上执行作业的 main() 方法。集群将关闭。
  
  作业提交示例：
```
flink run-application \
    -t yarn-application \
    -Djobmanager.memory.process.size=1024m \
    -Dtaskmanager.memory.process.size=2048m \
    -Djobmanager.heap.mb=820 \
    -Dtaskmanager.heap.mb=1640 \
    -Dtaskmanager.numberOfTaskSlots=2 \
    -Dparallelism.default=4 \
    /usr/lib/flink/examples/batch/WordCount.jar
```
  列出正在运行的作业：
```
./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
```
  取消正在运行的作业：
```
./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
```
2. 按作业模式：此 Flink 模式会在main() 客户端。
  
  作业提交示例：
```
flink run \
    -m yarn-cluster \
    -p 4 \
    -ys 2 \
    -yjm 1024m \
    -ytm 2048m \
    /usr/lib/flink/examples/batch/WordCount.jar
```
3. 会话模式：启动长时间运行的 Flink YARN 会话，然后提交一个或多个作业添加到会话中
  1. 启动会话：您可以在以下任一位置启动 Flink 会话：方法：
    1. 创建一个 Flink 集群，并添加 --metadata flink-start-yarn-session=true 标记添加到 gcloud dataproc clusters create 命令（请参阅创建一个 Dataproc Flink 集群。带有此标志创建集群后，Dataproc 将 /usr/bin/flink-yarn-daemon，用于在集群上启动 Flink 会话。
      
      会话的 YARN 应用 ID 保存在 /tmp/.yarn-properties-${USER} 中。您可以使用 yarn application -list 命令列出该 ID。
    2. 运行 Flink yarn-session.sh 脚本（预先安装在集群主服务器虚拟机上，具有自定义设置）：
      
      自定义设置示例：
```
/usr/lib/flink/bin/yarn-session.sh \
    -s 1 \
    -jm 1024m \
    -tm 2048m \
    -nm flink-dataproc \
    --detached
```
    3. 使用以下命令运行 Flink /usr/bin/flink-yarn-daemon 封装容器脚本默认设置：
```
. /usr/bin/flink-yarn-daemon
```
  2. 将作业提交到会话：运行以下命令以提交 Flink 作业与会话。
```
flink run -m <var>FLINK_MASTER_URL</var>/usr/lib/flink/examples/batch/WordCount.jar
```
    - FLINK_MASTER_URL：网址，包括主机和端口。 从以下位置移除 http:// prefix：。当您启动 Flink 会话。您可以运行以下命令列出此网址在 Tracking-URL 字段中：
```
yarn application -list -appId=<yarn-app-id> | sed 's#http://##'
   ```
```
3. 列出会话中的作业：如需列出会话中的 Flink 作业，请执行以下操作之一：以下：
  - 在不使用参数的情况下运行 flink list。该命令会查找 /tmp/.yarn-properties-${USER} 中会话的 YARN 应用 ID。
  - 从以下位置获取会话的 YARN 应用 ID： /tmp/.yarn-properties-${USER} 或 yarn application -list 的输出。然后运行 <code>flink list -yidYARN_APPLICATION_ID。
  - 运行 flink list -m FLINK_MASTER_URL。
4. 停止会话：如需停止会话，请获取 YARN 应用 ID 从 /tmp/.yarn-properties-${USER} 开始会话或 yarn application -list，然后运行以下任一命令：
```
  echo "stop" | /usr/lib/flink/bin/yarn-session.sh -id YARN_APPLICATION_ID
```
```
  yarn application -kill YARN_APPLICATION_ID
```

在 Flink 上运行 Apache Beam 作业

您可以使用 FlinkRunner 在 Dataproc 上运行 Apache Beam 作业。

您可以通过以下方式在 Flink 上运行 Beam 作业：

Java Beam 作业
可移植的 Beam 作业

Java Beam 作业

将 Beam 作业打包成一个 JAR 文件。为捆绑的 JAR 文件提供运行作业所需的依赖项。

以下示例从 Dataproc 集群的主节点运行 Java Beam 作业。

创建启用了 Flink 组件的 Dataproc 集群。
```
gcloud dataproc clusters create CLUSTER_NAME \
    --optional-components=FLINK \
    --image-version=DATAPROC_IMAGE_VERSION \
    --region=REGION \
    --enable-component-gateway \
    --scopes=https://www.googleapis.com/auth/cloud-platform
```
- --optional-components：Flink。
- --image-version：集群的映像版本，用于确定集群上安装的 Flink 版本（例如，请参阅针对最近 4 个 2.0.x 映像发布版本列出的 Apache Flink 组件版本）。
- --region：受支持的 Dataproc 区域。
- --enable-component-gateway：启用对 Flink Job Manager 界面的访问权限。
- --scopes：启用您的集群对 Google Cloud API 的访问权限（请参阅范围最佳实践）。 cloud-platform 范围默认处于启用状态（您不需要包含此标志设置）（使用 Dataproc 映像 2.1 或更高版本）。
使用 SSH 实用程序 在 Flink 集群主服务器节点上打开一个终端窗口。
在 Dataproc 集群主服务器实例上启动 Flink YARN 会话节点。
```
. /usr/bin/flink-yarn-daemon
```
记下 Dataproc 集群上的 Flink 版本。
```
flink --version
```
在本地机器上，使用 Java 生成规范的 Beam 字数统计示例。

选择与 Dataproc 集群上的 Flink 版本兼容的 Beam 版本。请参阅 Flink 版本兼容性列出了 Beam-Flink 版本兼容性的表

打开生成的 POM 文件。检查 <flink.artifact.name> 标记指定的 Beam Flink 运行程序版本。如果 Flink 工件名称中的 Beam Flink 运行程序版本与您的集群上的 Flink 版本不匹配，请更新版本号以使其相匹配。
```
mvn archetype:generate \
    -DarchetypeGroupId=org.apache.beam \
    -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
    -DarchetypeVersion=BEAM_VERSION \
    -DgroupId=org.example \
    -DartifactId=word-count-beam \
    -Dversion="0.1" \
    -Dpackage=org.apache.beam.examples \
    -DinteractiveMode=false
```
打包字数统计示例。
```
mvn package -Pflink-runner
```
将打包的超级 JAR 文件 word-count-beam-bundled-0.1.jar（约 135 MB）上传到您的 Dataproc 集群的主节点。您可以使用 gcloud storage cp 加快从 Cloud Storage 向 Dataproc 集群传输文件的速度。
1. 在本地终端上，创建一个 Cloud Storage 存储桶，然后上传超级 JAR。
```
gcloud storage buckets create BUCKET_NAME

gcloud storage cp target/word-count-beam-bundled-0.1.jar gs://BUCKET_NAME/
```
2. 在 Dataproc 的主节点上，下载超级 JAR。
```
gcloud storage cp gs://BUCKET_NAME/word-count-beam-bundled-0.1.jar .
```

在 Dataproc 集群的主节点上运行 Java Beam 作业。

flink run -c org.apache.beam.examples.WordCount word-count-beam-bundled-0.1.jar \
    --runner=FlinkRunner \
    --output=gs://BUCKET_NAME/java-wordcount-out

检查结果是否已写入 Cloud Storage 存储桶。

gcloud storage cat gs://BUCKET_NAME/java-wordcount-out-SHARD_ID

停止 Flink YARN 会话。

yarn application -list

yarn application -kill YARN_APPLICATION_ID

可移植的 Beam 作业

如需运行使用 Python、Go 和其他支持的语言编写的 Beam 作业，您可以按照 Beam 的FlinkRunnerPortableRunner Flink Runner 页面（另请参阅可移植性框架路线图）。

以下示例从 Dataproc 集群的主节点运行以 Python 编写的可移植 Beam 作业。

创建同时启用了 Flink 和 Docker 组件的 Dataproc 集群。
```
gcloud dataproc clusters create CLUSTER_NAME \
    --optional-components=FLINK,DOCKER \
    --image-version=DATAPROC_IMAGE_VERSION \
    --region=REGION \
    --enable-component-gateway \
    --scopes=https://www.googleapis.com/auth/cloud-platform
```
注意：
- --optional-components：Flink 和 Docker。
- --image-version：集群的映像版本。这决定了集群上安装的 Flink 版本（例如，请查看列出的 Apache Flink 组件版本，四个 2.0.x 映像发布版本）。
- --region：可用的 Dataproc 区域。
- --enable-component-gateway：启用对 Flink Job Manager 界面的访问权限。
- --scopes：启用您的集群对 Google Cloud API 的访问权限（请参阅范围最佳实践）。 cloud-platform 范围默认处于启用状态（您不需要包含此标志设置）（使用 Dataproc 映像 2.1 或更高版本）。
在本地或在 Cloud Shell，用于创建 Cloud Storage 存储桶。您将指定BUCKET_NAME 。
```
gcloud storage buckets create BUCKET_NAME
```
在集群虚拟机的终端窗口中，启动 Flink YARN 会话。记下 Flink 主网址，即 Flink 主实例的地址执行作业的位置。在执行以下操作时，您需要指定FLINK_MASTER_URL 运行示例 WordCount 程序。
```
. /usr/bin/flink-yarn-daemon
```
显示并记下运行 Dataproc 的 Flink 版本 集群。在执行以下操作时，您需要指定FLINK_VERSION 运行示例 WordCount 程序。
```
flink --version
```
在集群主服务器节点。

安装 Beam 版本兼容集群上的 Flink 版本。

python -m pip install apache-beam[gcp]==BEAM_VERSION

在集群主服务器实例上运行单词数示例节点。
```
python -m apache_beam.examples.wordcount \
    --runner=FlinkRunner \
    --flink_version=FLINK_VERSION \
    --flink_master=FLINK_MASTER_URL
    --flink_submit_uber_jar \
    --output=gs://BUCKET_NAME/python-wordcount-out
```
注意：
- --runner：FlinkRunner。
- --flink_version：之前提到的 FLINK_VERSION。
- --flink_master：之前提到的 FLINK_MASTER_URL。
- --flink_submit_uber_jar：使用超级 JAR 执行 Beam 作业。
- --output：BUCKET_NAME（之前创建）。

验证结果是否已写入存储桶。

gcloud storage cat gs://BUCKET_NAME/python-wordcount-out-SHARD_ID

停止 Flink YARN 会话。

获取应用 ID。

yarn application -list

1. Insert the <var>YARN_APPLICATION_ID</var>, then stop the session.

yarn application -kill

在 Kerberos 化的集群上运行 Flink

Dataproc Flink 组件支持 Kerberos 化的集群。需要有效的 Kerberos 票据才能提交和保留 Flink 作业，或启动一个 Flink 集群默认情况下，Kerberos 票据的有效期为 7 天。

访问 Flink Job Manager 界面

当 Flink 作业或 Flink 会话集群运行时，您可以使用 Flink Job Manager 网页界面。如需使用网页界面，请执行以下操作：

创建 Dataproc Flink 集群。
创建集群后，点击组件网关 集群详情页面上“网络界面”标签页上的 YARN ResourceManager 链接 Google Cloud 控制台中。
在 YARN Resource Manager 界面中，识别 Flink 集群应用条目。根据作业的完成状态，ApplicationMaster 或历史记录链接。
对于长时间运行的流处理作业，请点击 ApplicationManager 链接，以打开 Flink 信息中心；对于已完成的作业，请点击历史记录链接查看作业详情。

Dataproc 可选 Flink 组件

创建 Dataproc Flink 集群

控制台

gcloud

API

创建 Flink 集群后

使用 Dataproc Jobs 资源运行 Flink 作业

控制台

gcloud

REST

curl（Linux、macOS 或 Cloud Shell）

PowerShell (Windows)

使用 flink CLI 运行 Flink 作业

在 Flink 上运行 Apache Beam 作业

Java Beam 作业

可移植的 Beam 作业

在 Kerberos 化的集群上运行 Flink

访问 Flink Job Manager 界面

使用 Dataproc `Jobs` 资源运行 Flink 作业

使用 `flink` CLI 运行 Flink 作业