全部產品

：EMR Flink寫入OSS-HDFS服務

更新時間：Jun 19, 2024

可恢複性寫入功能支援將資料以EXACTLY_ONCE語義寫入儲存介質。本文介紹Flink如何通過EMR叢集的方式可恢複性寫入OSS-HDFS服務。

前提條件

已建立EMR-3.42.0及以上版本或EMR-5.8.0及以上版本的叢集。具體步驟，請參見建立叢集。
已開通並授權訪問OSS-HDFS服務。具體操作，請參見開通並授權訪問OSS-HDFS服務。

樣本

通用配置
為了支援EXACTLY_ONCE語義寫入OSS-HDFS，您需要執行如下配置：
1. 開啟Flink的檢查點（Checkpoint）。
  樣本如下。
  1. 通過如下方式建立的StreamExecutionEnvironment。
```
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
```
  2. 執行如下命令，啟動Checkpoint。
```
env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);
```
2. 使用可以重發的資料來源，例如Kafka。
便捷使用
您無需額外引入依賴，只需攜帶oss://首碼的路徑，並使用OSS-HDFS服務的Bucket及Endpoint，即可啟用Flink。
1. 添加Sink。
  以將DataStream<String>的對象OutputStream寫入OSS-HDFS為例。
```
String outputPath = "oss://<user-defined-oss-hdfs-bucket.oss-hdfs-endpoint>/<user-defined-dir>"
StreamingFileSink<String> sink = StreamingFileSink.forRowFormat(
        new Path(outputPath),
        new SimpleStringEncoder<String>("UTF-8")
).build();
outputStream.addSink(sink);
```
  重要
  在OSS-HDFS服務的Bucket中帶有.<oss-hdfs-endpoint>的欄位為可選項。如果您希望省略該欄位，請確保已在Flink或Hadoop組件中正確配置了OSS-HDFS服務的Endpoint。
2. 使用env.execute()執行Flink作業。

（可選）自訂配置

您在提交Flink作業時，可以自訂參數，以開啟或控制特定功能。

例如，通過-yD配置以yarn-cluster模式提交Flink作業時，樣本如下：

<flink_home>/bin/flink run -m yarn-cluster -yD key1=value1 -yD key2=value2 ...

您可以開啟熵注入（Entropy Injection）功能。熵注入可以匹配寫入路徑的一段特定字串，用一段隨機的字串進行替換，以削弱所謂片區效應，提高寫入效率。

當寫入情境為OSS-HDFS時，需要完成下列配置。

oss.entropy.key=<user-defined-key>
oss.entropy.length=<user-defined-length>

寫入新檔案時，路徑中與<user-defined-key>相同的字串會被替換為一個隨機字串，隨機串的長度為<user-defined-length>，且<user-defined-length>必須大於零。