Object Storage Service:Use Flink on an EMR cluster to write data to the OSS-HDFS service

Exemplo

1. Configurações gerais

   Para gravar dados no serviço OSS-HDFS com semântica EXACTLY_ONCE, conclua as seguintes configurações:

   1. Ative os checkpoints do Flink.

      Exemplo:

      1. Crie um objeto StreamExecutionEnvironment conforme o exemplo a seguir.

         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

      2. Execute o comando a seguir para ativar os checkpoints.

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

   2. Use uma source de dados que permita replay, como o Kafka.

2. Uso básico

   Não são necessárias dependências extras. Para gravar dados, use um caminho que comece com o prefixo oss:// e especifique o bucket e o endpoint do serviço OSS-HDFS.

   1. Adicione um sink.

      O exemplo a seguir mostra como gravar um objeto DataStream<String> no serviço 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);

      Importante

      A parte .{oss-hdfs-endpoint} do caminho é opcional. Se você omitir essa parte, configure corretamente o endpoint do serviço OSS-HDFS no componente Flink ou Hadoop.

   2. Use env.execute() para executar o job do Flink.

(Opcional) Defina configurações personalizadas

Defina parâmetros personalizados ao enviar jobs do Flink para ativar recursos específicos.

Use -yD para configurar o envio de jobs do Flink baseados em YARN:

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

A injeção de entropia substitui uma string especificada no caminho de destino por uma string aleatória para distribuir as gravações entre partições e melhorar o desempenho.

Configure os seguintes parâmetros para gravações no OSS-HDFS:

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

Durante as gravações, a string <user-defined-key> no caminho é substituída por uma string aleatória cujo comprimento é igual a <user-defined-length>. O valor de <user-defined-length> deve ser maior que 0.