用戶可將自己開發的程序提交到MRS中，執行程序并獲取結果。本章節指導用戶在MRS集群頁面如何提交一個新的Flink作業。Flink作業用于提交jar程序處理流式數據。
前提條件
用戶已經將運行作業所需的程序包和數據文件上傳至OBS系統或HDFS中。
通過界面提交作業
1.登錄MRS管理控制臺。
2.選擇“集群列表 > 現有集群”，選中一個運行中的集群并單擊集群名稱，進入集群信息頁面。
3.若集群開啟Kerberos認證時執行該步驟，若集群未開啟Kerberos認證，請無需執行該步驟。
在“概覽”頁簽的基本信息區域，單擊“IAM用戶同步”右側的“同步”進行IAM用戶同步。
說明
當IAM用戶的用戶組的所屬策略從MRS ReadOnlyAccess向MRS CommonOperations、MRS FullAccess、MRS Administrator變化時，由于集群節點的SSSD（System Security Services Daemon）緩存刷新需要時間，因此同步完成后，請等待5分鐘，等待新修改策略生效之后，再進行提交作業。否則，會出現提交作業失敗的情況。
當IAM用戶的用戶組的所屬策略從MRS CommonOperations、MRS FullAccess、MRS Administrator向MRS ReadOnlyAccess變化時，由于集群節點的SSSD緩存刷新需要時間，因此同步完成后，請等待5分鐘，新修改策略才能生效。
4.單擊“作業管理”，進入“作業管理”頁簽。
5.單擊“添加”，進入“添加作業”頁面。
6.“作業類型”選擇“Flink”，參考下表配置Flink作業信息。
作業配置信息
| 參數 | 參數說明 |
|---|---|
| 作業名稱 | 作業名稱，只能由字母、數字、中劃線和下劃線組成，并且長度為1～64個字符。 說明 建議不同的作業設置不同的名稱。 |
| 執行程序路徑 | 待執行程序包地址，需要滿足如下要求： 最多為1023字符，不能包含;l&>,<'$特殊字符，且不可為空或全空格。 執行程序路徑可存儲于HDFS或者OBS中，不同的文件系統對應的路徑存在差異。 - OBS：以“obs://”開頭。示例： obs://wordcount/program/xxx.jar。 - HDFS：以“/user”開頭。數據導入HDFS請參考導入導出數據章節中的“導入數據”。 |
| 運行程序參數 | 可選參數，為本次執行的作業配置相關優化參數（例如線程、內存、CPU核數等），用于優化資源使用效率，提升作業的執行性能。 常用運行程序參數如下表“運行程序參數”。 |
| 執行程序參數 | 可選參數，程序執行的關鍵參數，該參數由用戶程序內的函數指定，MRS只負責參數的傳入。多個參數間使用空格隔開。 最多為150000字符，不能包含不能包含;;&><'$特殊字符，可為空。 注意 若輸入帶有敏感信息（如登錄密碼）的參數可能在作業詳情展示和日志打印中存在暴露的風險，請謹慎操作。 |
| 服務配置參數 | 可選參數，用于為本次執行的作業修改服務配置參數。該參數的修改僅適用于本次執行的作業，如需對集群永久生效，請參考配置服務參數頁面進行修改。 如需添加多個參數，請單擊右側   增加，如需刪除參數，請單擊右側“刪除”。 常用服務配置參數如下表“服務配置參數”。 |
| 命令參考 | 用于展示提交作業時提交到后臺執行的命令。 |
運行程序參數
| 參數 | 參數說明 | 取值樣例 |
|---|---|---|
| -ytm | 設置每個TaskManager容器的內存（單位可選,默認單位：MB）。 | 1024 |
| -yjm | 設置JobManager容器內存（單位可選，默認單位：MB）。 | 1024 |
| -yn | 設置分配給應用程序的Yarn容器的數量，該值與TaskManager數量相同。 | 2 |
| -ys | 設置TaskManager的核數。 | 2 |
| -ynm | 自定義Yarn上應用程序名稱。 | test |
| -c | 設置程序入口點的類（如“main”或“getPlan(）”方法）。該參數僅在JAR文件未指定其清單的類時需要。 | com.bigdata.mrs.test |
說明針對MRS 3.x及之后版本，運行程序參數不支持“-yn”。
服務配置參數
| 參數 | 參數說明 | 取值樣例 |
|---|---|---|
| fs.obs.access.key | 訪問OBS的密鑰ID。 | - |
| fs.obs.secret.key | 訪問OBS與密鑰ID對應的密鑰。 | - |
7.確認作業配置信息，單擊“確定”，完成作業的新增。
作業新增完成后，可對作業進行管理。
通過后臺提交作業
MRS 3.x及之后版本客戶端默認安裝路徑為“/opt/Bigdata/client”，MRS 3.x之前版本為“/opt/client”。具體以實際為準。
1.登錄MRS客戶端。
2.執行如下命令初始化環境變量。
source /opt/Bigdata/client/bigdata_env
3.若集群開啟Kerberos認證，需要執行以下步驟，若集群未開啟Kerberos認證請跳過該步驟。
a.準備一個提交Flink作業的用戶。
b.使用新創建的用戶登錄Manager頁面。
- MRS 3.x之前版本，登錄集群的Manager界面，選擇“系統設置 > 用戶管理”，在已增加用戶所在行的“操作”列，選擇“更多 > 下載認證憑據”。
- MRS 3.x及之后版本，登錄集群的Manager界面，選擇“系統 > 權限 > 用戶”，在已增加用戶所在行的“操作”列，選擇“更多 > 下載認證憑據”。
c.將下載的認證憑據壓縮包解壓縮，并將得到的user.keytab文件拷貝到客戶端節點中，例如客戶端節點的“/opt/Bigdata/client/Flink/flink/conf”目錄下。如果是在集群外節點安裝的客戶端，需要將得到的krb5.conf文件拷貝到該節點的“/etc/”目錄下。
d.MRS 3.x及之后版本，安全模式下需要將客戶端安裝節點的業務IP以及Manager的浮動ip追加到“/opt/Bigdata/client/Flink/flink/conf/flink-conf.yaml”文件中的“jobmanager.web.allow-access-address”配置項中，ip之間使用英文逗號分隔。
e.配置安全認證，在“/opt/Bigdata/client/Flink/flink/conf/flink-conf.yaml”配置文件中的對應配置添加keytab路徑以及用戶名。
security.kerberos.login.keytab: <user.keytab 文件路徑 >
security.kerberos.login.principal: < 用戶名 >
例如：
security.kerberos.login.keytab: /opt/Bigdata/client/Flink/flink/conf/user.keytab
security.kerberos.login.principal: test
f.在Flink的客戶端bin目錄下，執行如下命令進行安全加固，password請重新設置為一個用于提交作業的密碼。
sh generate_keystore.sh < password >
該腳本會自動替換“/opt/Bigdata/client/Flink/flink/conf/flink-conf.yaml”中關于SSL的值，針對MRS 3.x之前版本，安全集群默認沒有開啟外部SSL，用戶如果需要啟用外部SSL，進行配置后再次運行該腳本即可，配置參數在MRS的Flink默認配置中不存在，用戶如果開啟外部連接SSL，則需要添加下表中參數。
參數描述
| 參數 | 參數值示例 | 描述 |
|---|---|---|
| security.ssl.rest.enabled | true | 打開外部SSL開關。 |
| security.ssl.rest.keystore | ${path}/flink.keystore | keystore的存放路徑。 |
| security.ssl.rest.keystore-password | 123456 | keystore的password，“123456”表示需要用戶輸入自定義設置的密碼值。 |
| security.ssl.rest.key-password | 123456 | ssl key的password，“123456”表示需要用戶輸入自定義設置的密碼值。 |
| security.ssl.rest.truststore | ${path}/flink.truststore | truststore存放路徑。 |
| security.ssl.rest.truststore-password | 123456 | truststore的password，“123456”表示需要用戶輸入自定義設置的密碼值。 |
說明
針對MRS 3.x之前版本，generate_keystore.sh腳本無需手動生成。
會將生成的flink.keystore、flink.truststore、security.cookie自動填充到“flink-conf.yaml”對應配置項中。
針對MRS 3.x及之后版本，“security.ssl.key-password”、“security.ssl.keystore-password”和“security.ssl.truststore-password”的值需要使用Manager明文加密API進行獲取：
curl -k -i -u < user name >:< password > -X POST -HContent-type:application/json -d '{"plainText":"< password >"}' '// x.x.x.x :28443/web/api/v2/tools/encrypt' ；其中< password >要與簽發證書時使用的密碼一致，x.x.x.x為集群Manager的浮動IP。
g.客戶端訪問flink.keystore和flink.truststore文件的路徑配置。
- 絕對路徑：執行該腳本后，在flink-conf.yaml文件中將flink.keystore和flink.truststore文件路徑自動配置為絕對路徑“/opt/Bigdata/client/Flink/flink/conf/”，此時需要將conf目錄中的flink.keystore和flink.truststore文件分別放置在Flink Client以及Yarn各個節點的該絕對路徑上。
- 相對路徑：請執行如下步驟配置flink.keystore和flink.truststore文件路徑為相對路徑，并確保Flink Client執行命令的目錄可以直接訪問該相對路徑。
i.在“/opt/Bigdata/client/Flink/flink/conf/”目錄下新建目錄，例如ssl。
ii. 移動flink.keystore和flink.truststore文件到“/opt/Bigdata/client/Flink/flink/conf/ssl/”中。
iii. 針對MRS 3.x及之后版本，修改flink-conf.yaml文件中如下兩個參數為相對路徑。
security.ssl.keystore: ssl/flink.keystore
security.ssl.truststore: ssl/flink.truststore
iv. 針對MRS 3.x之前版本，修改flink-conf.yaml文件中如下兩個參數為相對路徑。
security.ssl.internal.keystore: ssl/flink.keystore
security.ssl.internal.truststore: ssl/flink.truststore
h.如果客戶端安裝在集群外節點，請在配置文件（如：“ /opt/Bigdata/client/Flink/fink/conf/flink-conf.yaml ”） 中增加如下配置值，其中xx.xx.xxx.xxx請替換為客戶端所在節點的IP。
web.access-control-allow-origin: xx.xx.xxx.xxx
jobmanager.web.allow-access-address: xx.xx.xxx.xxx
4.運行wordcount作業。
- 普通集群（未開啟Kerberos認證）
?執行如下命令啟動session，并在session中提交作業。
yarn-session.sh -nm "session-name"
flink run /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
?執行如下命令在Yarn上提交單個作業。
flink run -m yarn-cluster /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
- 安全集群（開啟Kerberos認證）
?flink.keystore和flink.truststore文件路徑為絕對路徑時：
執行如下命令啟動session，并在session中提交作業。
yarn-session.sh -nm "session-name"
flink run /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
執行如下命令在Yarn上提交單個作業。
flink run -m yarn-cluster /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
?flink.keystore和flink.truststore文件路徑為相對路徑時：
在“ssl”的同級目錄下執行如下命令啟動session，并在session中提交作業，其中“ssl”是相對路徑，如“ssl”所在目錄是“opt/Bigdata/client/Flink/flink/conf/”，則在“opt/Bigdata/client/Flink/flink/conf/”目錄下執行命令。
yarn-session.sh -t ssl/ -nm "session-name"
flink run /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
執行如下命令在Yarn上提交單個作業。
flink run -m yarn-cluster -yt ssl/ /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
