前提条件

• 翼MR集群中已安装Flink组件。
• 集群正常运行，已安装集群客户端，例如安装目录为“/usr/local/flink”。以下操作的客户端目录只是举例，请根据实际安装目录修改。

使用Flink客户端

1. 安装客户端，具体请参考安装客户端章节。

2. 登录安装客户端的节点。

3. 执行以下命令，切换到客户端安装目录。

   cd /opt/hadoopclient

4. 执行如下命令初始化环境变量。

   source /opt/hadoopclient/bigdata_env 
   source ~/.bash_profile
   

5. 若集群开启Kerberos认证，需要执行以下命令以完成认证，若集群未开启Kerberos认证请跳过该步骤。

   klist -kt /etc/security/keytabs/hdfs.keytab
   kinit -kt /etc/security/keytabs/hdfs.keytab XXXX
   

   同时，应配置安全认证。在“/usr/local/flink/conf/flink-conf.yaml”配置文件中的对应配置添加keytab路径以及用户名。

   security.kerberos.login.use-ticket-cache: true
   security.kerberos.login.keytab: <user.keytab 文件路径 >
   security.kerberos.login.principal: user
   security.kerberossecurity.login.contexts: Client,KafkaClient
   

   例如：

   security.kerberos.login.keytab: /etc/security/keytabs/hdfs.keytab
   security.kerberos.login.principal: hdfs
   

6. 运行wordcount作业。

   1. 普通集群（未开启Kerberos认证）

      首先启动Flink集群。

      /usr/local/flink/bin/start-cluster.sh
      Session模式

      执行如下命令在session中提交作业。
      yarn-session.sh -nm "session-name" --detachedflink run /usr/local/flink/examples/streaming/WordCount.jar
      提交成功后，会返回已提交的Flink作业的YARN Application ID以及Web地址，访问Web地址以通过Web UI的方式查看作业状态。
      Per-Job模式
      执行如下命令以Per-Job方式提交作业。
      flink run -t yarn-per-job --detached /usr/local/flink/examples/streaming/WordCount.jar
      提交成功后，会返回已提交的Flink作业的YARN Application ID以及Web地址，访问Web地址以通过Web UI的方式查看作业状态。
      Application模式
      执行如下命令以Application方式提交作业。
      flink run-application -t yarn-application/usr/local/flink/examples/streaming/WordCount.jar
      提交成功后，会返回已提交的Flink作业的YARN Application ID以及Web地址，访问Web地址以通过Web UI的方式查看作业状态。

   2. 安全集群（开启Kerberos认证）
      首先执行5中的认证操作，再进行和普通集群相同的操作以运行作业。