可恢複性寫入功能支援將資料以EXACTLY_ONCE語義寫入儲存介質。本文介紹Flink如何通過EMR叢集的方式可恢複性寫入OSS-HDFS服務。
前提條件
已為Bucket開通並授權訪問OSS-HDFS服務。具體操作,請參見開通OSS-HDFS服務。
阿里雲帳號預設擁有通過EMR叢集的方式接入OSS-HDFS服務並執行常見操作的許可權。如果您希望通過RAM使用者接入OSS-HDFS服務,RAM使用者需要具備對應的許可權要求,詳情請參見授權RAM使用者通過EMR叢集接入OSS-HDFS服務。
樣本
通用配置
為了支援EXACTLY_ONCE語義寫入OSS-HDFS,您需要執行如下配置:
開啟Flink的檢查點(Checkpoint)。
樣本如下。
通過如下方式建立的StreamExecutionEnvironment。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();執行如下命令,啟動Checkpoint。
env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);
使用可以重發的資料來源,例如Kafka。
便捷使用
您無需額外引入依賴,只需攜帶oss://首碼的路徑,並使用OSS-HDFS服務的Bucket及Endpoint,即可啟用Flink。
添加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。使用
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>必須大於零。