支持Flink可恢复性写入JindoFS或OSS

在Flink作业中的用法

• 通用配置
  为了支持EXACTLY_ONCE语义写入JindoFS或OSS，您需要执行如下配置：

  1. 打开Flink的检查点（Checkpoint）。
     示例如下。

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

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

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

  您无需额外引入依赖，只需使用带有jfs://或oss://前缀的路径，就可以使用该功能。JindoFS可以自动识别jfs://或oss://前缀，并启用该功能。

  例如，以DataStream<String>的对象OutputStream为例。

  1. 添加Sink。
     • 将其写入JindoFS时，您可以执行如下命令添加Sink。

       String outputPath = "jfs://<user-defined-jfs-namespace>/<user-defined-jfs-dir>"
       StreamingFileSink<String> sink = StreamingFileSink.forRowFormat(
               new Path(outputPath),
               new SimpleStringEncoder<String>("UTF-8")
       ).build();
       outputStream.addSink(sink);

     • 将其写入OSS，您可以执行如下命令添加Sink。

       String outputPath = "oss://<user-defined-oss-bucket>/<user-defined-oss-dir>"
       StreamingFileSink<String> sink = StreamingFileSink.forRowFormat(
               new Path(outputPath),
               new SimpleStringEncoder<String>("UTF-8")
       ).build();
       outputStream.addSink(sink);

  2. 使用env.execute()执行Flink作业即可。

自定义配置

您在提交Flink作业时，可以自定义参数，以开启或控制特定功能。

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