使用Flink客户端 本章节主要介绍使用Flink运行wordcount作业的操作指导。 前提条件 MRS集群中已安装Flink组件。 集群正常运行,已安装集群客户端,例如安装目录为“/opt/hadoopclient”。以下操作的客户端目录只是举例,请根据实际安装目录修改。 使用Flink客户端(MRS 3.x之前版本) 1. 以客户端安装用户,登录安装客户端的节点。 2. 执行以下命令,切换到客户端安装目录。 cd /opt/hadoopclient 3. 执行如下命令初始化环境变量。 source /opt/hadoopclient/bigdataenv 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/flinkconf.yaml”配置文件中的对应配置添加keytab路径以及用户名。 security.kerberos.login.keytab: security.kerberos.login.principal: 例如: security.kerberos.login.keytab:/opt/hadoopclient/Flink/flink/conf/user.keytab security.kerberos.login.principal: test e.参考“组件操作指南 > 使用Flink > 参考 > 签发证书样例”章节生成“generatekeystore.sh”脚本并放置在Flink的客户端bin目录下,执行如下命令进行安全加固,请参考“组件操作指南 > 使用Flink > 安全加固 > 认证和加密”,password请重新设置为一个用于提交作业的密码。 sh generatekeystore.sh 该脚本会自动替换“/opt/hadoopclient/Flink/flink/conf/flinkconf.yaml”中关于SSL的值,针对MRS2.x及之前版本,安全集群默认没有开启外部SSL,用户如果需要启用外部SSL,请参考“组件操作指南 >使用Flink > 安全加固”进行配置后再次运行该脚本即可。 说明 generatekeystore.sh脚本无需手动生成。 执行认证和加密后会将生成的flink.keystore、flink.truststore、security.cookie自动填充到“flinkconf.yaml”对应配置项中。 f.客户端访问flink.keystore和flink.truststore文件的路径配置。 −绝对路径:执行该脚本后,在flinkconf.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. 修改flinkconf.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中提交作业。 yarnsession.sh nm " sessionname " flink run/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar −执行如下命令在Yarn上提交单个作业。 flink run m yarncluster /opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar 安全集群(开启Kerberos认证) −flink.keystore和flink.truststore文件路径为绝对路径时: 执行如下命令启动session,并在session中提交作业。 yarnsession.sh nm " sessionname " flink run /opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar 执行如下命令在Yarn上提交单个作业。 flink run m yarncluster /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/”目录下执行命令。 yarnsession.sh t ssl/ nm " sessionname " flink run/opt/hadoopclient/Flink/flink/examples/streaming/WordCount.jar 执行如下命令在Yarn上提交单个作业。 flink run m yarncluster 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