Write streaming data to OSS-HDFS with JindoSDK and Apache Flink - Object Storage Service

Open source Apache Flink does not support streaming writes to OSS-HDFS or exactly-once semantics for storage media. To enable these capabilities, use JindoSDK with Apache Flink.

Note

To write streaming data to OSS-HDFS without deploying JindoSDK, use Realtime Compute for Apache Flink. For more information, see Use Realtime Compute for Apache Flink to read from or write to OSS or OSS-HDFS.

Prerequisites

An ECS instance is created. Create an instance.
OSS-HDFS is enabled for a bucket with appropriate access permissions granted. For more information, see Enable OSS-HDFS.
Apache Flink 1.10.1 or later is installed. Flink 1.16.0 and later versions are not verified. Download Flink from Apache Flink.

Configure JindoSDK

Log on to the ECS instance. Connect to an instance.
Download and decompress the latest version of the JindoSDK JAR package. To download JindoSDK, visit GitHub.
Move jindo-flink-${version}-full.jar from the JindoSDK plugins/flink/ directory to the Apache Flink root lib directory.
```
mv plugins/flink/jindo-flink-${version}-full.jar lib/
```

Important

If Apache Flink has a built-in OSS connector, remove it by deleting flink-oss-fs-hadoop-${flink-version}.jar from the lib subdirectory or plugins/oss-fs-hadoop path.
After you configure JindoSDK, use the oss:// prefix in Flink jobs to write streaming data to OSS-HDFS or OSS. JindoSDK automatically identifies the target storage.

Examples

Configure general settings.

To write data to OSS-HDFS with exactly-once semantics:
1. Enable Apache Flink checkpointing.
  Sample code:
  1. Create a StreamExecutionEnvironment:
```
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
```
  2. Enable checkpointing:
```
env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);
```
2. Use a data source that supports data retransmission, such as Kafka.
Configure quick settings.

Specify a path that starts with oss:// with OSS-HDFS buckets and endpoints. No additional dependencies are required.
1. Add a sink.
  
  The following code writes a DataStream<String> to 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);
```
  Important
  The .{oss-hdfs-endpoint} field is optional. If omitted, specify the correct OSS-HDFS endpoint in the Flink or Hadoop configuration.
2. Run env.execute() to submit the Flink job.

(Optional) Configure custom settings

You can configure custom parameters when submitting Flink jobs to enable specific features.

Use -yD to configure YARN-based Flink job submissions:

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

Entropy injection replaces a specified string in the destination path with a random string to distribute writes across partitions and improve performance.

Configure the following parameters for OSS-HDFS writes:

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

During writes, the <user-defined-key> string in the path is replaced by a random string whose length equals <user-defined-length>. The value of <user-defined-length> must be greater than 0.