EMR Flink写入OSS-HDFS服务

示例

1. 通用配置

   为了支持EXACTLY_ONCE语义写入OSS-HDFS，您需要执行如下配置：

   1. 打开Flink的检查点（Checkpoint）。

      示例如下。

      1. 通过如下方式建立的StreamExecutionEnvironment。

         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

      2. 执行如下命令，启动Checkpoint。

         env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);

   2. 使用可以重发的数据源，例如Kafka。

2. 便捷使用

   您无需额外引入依赖，只需携带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>必须大于零。