前提条件
- MRS集群中已安装Flink组件。
- 集群正常运行，已安装集群客户端，例如安装目录为“/opt/hadoopclient”。以下操作的客户端目录只是举例，请根据实际安装目录修改。
使用Flink客户端（MRS 3.x之前版本）
- 以客户端安装用户，登录安装客户端的节点。
- 执行以下命令，切换到客户端安装目录。
cd /opt/hadoopclient
- 执行如下命令初始化环境变量。
source /opt/hadoopclient/bigdata_env
4.若集群开启Kerberos认证，需要执行以下步骤，若集群未开启Kerberos认证请跳过该步骤。
a.准备一个提交Flink作业的用户。
b.登录Manager，下载认证凭据。
登录集群的Manager界面，具体请参见访问MRSManager（MRS2.x及之前版本），选择“系统设置 > 用户管理”，在已增加用户所在行的“操作”列，选择“更多 > 下载认证凭据”。
c.将下载的认证凭据压缩包解压缩，并将得到的user.keytab文件拷贝到客户端节点中，例如客户端节点的“/opt/hadoopclient/Flink/flink/conf”目录下。如果是在集群外节点安装的客户端，需要将得到的krb5.conf文件拷贝到该节点的“/etc/”目录下。
d.配置安全认证，在“/opt/hadoopclient/Flink/flink/conf/flink-conf.yaml”配置文件中的对应配置添加keytab路径以及用户名。
security.kerberos.login.keytab: <user.keytab 文件路径 >
security.kerberos.login.principal: < 用户名 >
例如：
security.kerberos.login.keytab:/opt/hadoopclient/Flink/flink/conf/user.keytab
security.kerberos.login.principal: test
e.参考“组件操作指南 > 使用Flink > 参考 > 签发证书样例”章节生成“generate_keystore.sh”脚本并放置在Flink的客户端bin目录下，执行如下命令进行安全加固，请参考“组件操作指南 > 使用Flink > 安全加固 > 认证和加密”，password请重新设置为一个用于提交作业的密码。
sh generate_keystore.sh < password >
该脚本会自动替换“/opt/hadoopclient/Flink/flink/conf/flink-conf.yaml”中关于SSL的值，针对MRS2.x及之前版本，安全集群默认没有开启外部SSL，用户如果需要启用外部SSL，请参考“组件操作指南 >使用Flink > 安全加固”进行配置后再次运行该脚本即可。
说明
generate_keystore.sh脚本无需手动生成。
执行认证和加密后会将生成的flink.keystore、flink.truststore、security.cookie自动填充到“flink-conf.yaml”对应配置项中。
f.客户端访问flink.keystore和flink.truststore文件的路径配置。
−绝对路径：执行该脚本后，在flink-conf.yaml文件中将flink.keystore和flink.truststore文件路径自动配置为绝对路径“/opt/hadoopclient/Flink/flink/conf/”，此时需要将conf目录中的flink.keystore和flink.truststore文件分别放置在Flink Client以及Yarn各个节点的该绝对路径上。
−相对路径：请执行如下步骤配置flink.keystore和flink.truststore文件路径为相对路径，并确保Flink Client执行命令的目录可以直接访问该相对路径。
i.在“/opt/hadoopclient/Flink/flink/conf/”目录下新建目录，例如ssl。
cd /opt/hadoopclient/Flink/flink/conf/
mkdir ssl
ii. 移动flink.keystore和flink.truststore文件到“/opt/hadoopclient/Flink/flink/conf/ssl/”中。
mv flink.keystore ssl/
mv flink.truststore ssl/
iii. 修改flink-conf.yaml文件中如下两个参数为相对路径。
security.ssl.internal.keystore: ssl/flink.keystore
security.ssl.internal.truststore: ssl/flink.truststore
5.运行wordcount作业。
须知
用户在Flink提交作业或者运行作业时，应具有如下权限：
- 如果启用Ranger鉴权，当前用户必须属于hadoop组或者已在Ranger中为该用户添加“/flink”的读写权限。
- 如果停用Ranger鉴权，当前用户必须属于hadoop组。
- 普通集群（未开启Kerberos认证）
−执行如下命令启动session，并在session中提交作业。
yarn-session.sh -nm " session-name "
flink run/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
−执行如下命令在Yarn上提交单个作业。
flink run -m yarn-cluster
/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 安全集群（开启Kerberos认证）
−flink.keystore和flink.truststore文件路径为绝对路径时：
- 执行如下命令启动session，并在session中提交作业。
yarn-session.sh -nm " session-name "
flink run /opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 执行如下命令在Yarn上提交单个作业。
flink run -m yarn-cluster
/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
−flink.keystore和flink.truststore文件路径为相对路径时：
- 在“ssl”的同级目录下执行如下命令启动session，并在session中提交作业，其中“ssl”是相对路径，如“ssl”所在目录是“opt/hadoopclient/Flink/flink/conf/”，则在“opt/hadoopclient/Flink/flink/conf/”目录下执行命令。
yarn-session.sh -t ssl/ -nm " session-name "
flink run/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 执行如下命令在Yarn上提交单个作业。
flink run -m yarn-cluster -yt ssl/ /opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
6.作业提交成功后，客户端界面显示如下。
详见下图： 在Yarn上提交作业成功
详见下图：启动session成功
详见下图： 在session中提交作业成功
7. 使用运行用户进入Yarn服务的原生页面，具体操作参考“组件操作指南 > 使用Flink > 查看Flink作业”，找到对应作业的application，单击application名称，进入到作业详情页面。
- 若作业尚未结束，可单击“Tracking URL”链接进入到Flink的原生页面，查看作业的运行信息。
- 若作业已运行结束，对于在session中提交的作业，可以单击“Tracking URL”链接登录Flink原生页面查看作业信息。
详见下图：application
使用Flink客户端（MRS 3.x及之后版本）
1.以客户端安装用户，登录安装客户端的节点。
2.执行以下命令，切换到客户端安装目录。
cd /opt/hadoopclient
3.执行如下命令初始化环境变量。
source/opt/hadoopclient/bigdata_env
4.若集群开启Kerberos认证，需要执行以下步骤，若集群未开启Kerberos认证请跳过该步骤。
a.准备一个提交Flink作业的用户。
b.登录Manager，下载认证凭据。
登录集群的Manager界面，具体请参见访问FusionInsight Manager（MRS 3.x及之后版本），选择“系统 > 权限 > 用户”，在已增加用户所在行的“操作”列，选择“更多 > 下载认证凭据”。
c.将下载的认证凭据压缩包解压缩，并将得到的user.keytab文件拷贝到客户端节点中，例如客户端节点的“/opt/hadoopclient/Flink/flink/conf”目录下。如果是在集群外节点安装的客户端，需要将得到的krb5.conf文件拷贝到该节点的“/etc/”目录下。
d.将客户端安装节点的业务IP、Manager的浮动IP和Master节点IP添加到配置文件“/opt/hadoopclient/Flink/flink/conf/flink-conf.yaml”中的“jobmanager.web.access-control-allow-origin”和“jobmanager.web.allow-access-address”配置项中，IP地址之间使用英文逗号分隔。
jobmanager.web.access-control-allow-origin:
xx.xx.xxx.xxx , xx.xx.xxx.xxx ,xx.xx.xxx.xxx
jobmanager.web.allow-access-address: xx.xx.xxx.xxx , xx.xx.xxx.xxx ,xx.xx.xxx.xxx
说明
客户端安装节点的业务IP获取方法：
集群内节点：
登录MapReduce服务管理控制台，选择“集群列表 > 现有集群”，选中当前的集群并单击集群名，进入集群信息页面。
在“节点管理”中查看安装客户端所在的节点IP。
- 集群外节点：安装客户端所在的弹性云主机的IP。
- Manager的浮动IP获取方法：
- 登录MapReduce服务管理控制台，选择“集群列表 > 现有集群”，选中当前的集群并单击集群名，进入集群信息页面。
在“节点管理”中查看节点名称，名称中包含“master1”的节点为Master1节点，名称中包含“master2”的节点为Master2节点。
- 远程登录Master2节点，执行“ifconfig”命令，系统回显中“eth0:wsom”表示MRS Manager浮动IP地址，请记录“inet”的实际参数值。如果在Master2节点无法查询到MRS Manager的浮动IP地址，请切换到Master1节点查询并记录。如果只有一个Master节点时，直接在该Master节点查询并记录。
e.配置安全认证，在“/opt/hadoopclient/Flink/flink/conf/flink-conf.yaml”配置文件中的对应配置添加keytab路径以及用户名。
security.kerberos.login.keytab: <user.keytab 文件路径 >
security.kerberos.login.principal: < 用户名 >
例如：
security.kerberos.login.keytab:
/opt/hadoopclient/Flink/flink/conf/user.keytab
security.kerberos.login.principal: test
f.参考“组件操作指南 > 使用Flink > 参考 > 签发证书样例”章节生成“generate_keystore.sh”脚本并放置在Flink的客户端bin目录下，执行如下命令进行安全加固，请参考“组件操作指南 >使用Flink > 安全加固 > 认证和加密”，password请重新设置为一个用于提交作业的密码。
sh generate_keystore.sh < password >
该脚本会自动替换“/opt/hadoopclient/Flink/flink/conf/flink-conf.yaml”中关于SSL的值。
sh generate_keystore.sh < password >
说明执行认证和加密后会在Flink客户端的“conf”目录下生成“flink.keystore”和“flink.truststore”文件，并且在客户端配置文件“flink-conf.yaml”中将以下配置项进行了默认赋值：
将配置项“security.ssl.keystore”设置为“flink.keystore”文件所在绝对路径。
将配置项“security.ssl.truststore”设置为“flink.truststore”文件所在的绝对路径。
将配置项“security.cookie”设置为“generate_keystore.sh”脚本自动生成的一串随机规则密码。
默认“flink-conf.yaml”中“security.ssl.encrypt.enabled: false”，“generate_keystore.sh”脚本将配置项“security.ssl.key-password”、“security.ssl.keystore-password”和“security.ssl.truststore-password”的值设置为调用“generate_keystore.sh”脚本时输入的密码。
g.客户端访问flink.keystore和flink.truststore文件的路径配置。
−绝对路径：执行该脚本后，在flink-conf.yaml文件中将flink.keystore和flink.truststore文件路径自动配置为绝对路径“/opt/hadoopclient/Flink/flink/conf/”，此时需要将conf目录中的flink.keystore和flink.truststore文件分别放置在Flink Client以及Yarn各个节点的该绝对路径上。
−相对路径：请执行如下步骤配置flink.keystore和flink.truststore文件路径为相对路径，并确保Flink Client执行命令的目录可以直接访问该相对路径。
i.在“/opt/hadoopclient/Flink/flink/conf/”目录下新建目录，例如ssl。
cd /opt/hadoopclient/Flink/flink/conf/
mkdir ssl
ii. 移动flink.keystore和flink.truststore文件到“/opt/hadoopclient/Flink/flink/conf/ssl/”中。
mv flink.keystore ssl/
mv flink.truststore ssl/
iii. 修改flink-conf.yaml文件中如下两个参数为相对路径。
security.ssl.keystore: ssl/flink.keystore
security.ssl.truststore: ssl/flink.truststore
5.运行wordcount作业。
须知
用户在Flink提交作业或者运行作业时，应具有如下权限：
- 如果启用Ranger鉴权，当前用户必须属于hadoop组或者已在Ranger中为该用户添加“/flink”的读写权限。
- 如果停用Ranger鉴权，当前用户必须属于hadoop组。
- 普通集群（未开启Kerberos认证）
−执行如下命令启动session，并在session中提交作业。
yarn-session.sh -nm " session-name "
flink run/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
−执行如下命令在Yarn上提交单个作业。
flink run -m yarn-cluster/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 安全集群（开启Kerberos认证）
−flink.keystore和flink.truststore文件路径为绝对路径时：
- 执行如下命令启动session，并在session中提交作业。
yarn-session.sh -nm " session-name "
flink run
/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 执行如下命令在Yarn上提交单个作业。
flink run -m yarn-cluster /opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
−flink.keystore和flink.truststore文件路径为相对路径时：
- 在“ssl”的同级目录下执行如下命令启动session，并在session中提交作业，其中“ssl”是相对路径，如“ssl”所在目录是“opt/hadoopclient/Flink/flink/conf/”，则在“opt/hadoopclient/Flink/flink/conf/”目录下执行命令。
yarn-session.sh -t ssl/ -nm " session-name "
flink run/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 执行如下命令在Yarn上提交单个作业。
flink run -m yarn-cluster -yt ssl/
/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar
- 作业提交成功后，客户端界面显示如下。
详见下图：在Yarn上提交作业成功
详见下图：启动session成功
详见下图：在session中提交作业成功
- 使用运行用户进入Yarn服务的原生页面，具体操作参考“组件操作指南 >使用Flink > 查看Flink作业”，找到对应作业的application，单击application名称，进入到作业详情页面
- 若作业尚未结束，可单击“Tracking URL”链接进入到Flink的原生页面，查看作业的运行信息。
- 若作业已运行结束，对于在session中提交的作业，可以单击“Tracking URL”链接登录Flink原生页面查看作业信息。
详见下图： application
