Write streaming data to OSS-HDFS with JindoSDK and Apache Flink - Object Storage Service: Central de Documentação do Alibaba Cloud

O Apache Flink de código aberto não oferece suporte a gravação em streaming no OSS-HDFS nem à semântica exactly-once para meios de armazenamento. Para ative esses recursos, use o JindoSDK com o Apache Flink.

Nota

Para gravar dados em streaming no OSS-HDFS sem implantar o JindoSDK, use o Realtime Compute for Apache Flink. Para obter mais informações, consulte 实时计算Flink读写OSS或者OSS-HDFS.

Pré-requisitos

crie uma instância do ECS. Create an instance.
ative o OSS-HDFS para um bucket e conceda as permissões de acesso adequadas. Para obter mais informações, consulte Enable OSS-HDFS service.
Instale o Apache Flink 1.10.1 ou posterior. As versões 1.16.0 e superiores do Flink não foram validadas. baixe o Flink em Apache Flink.

configure o JindoSDK

Conecte-se à instância do ECS. Connect to an instance.
baixe e descompacte a versão mais recente do pacote JAR do JindoSDK. Para baixe o JindoSDK, acesse GitHub.
Mova o arquivo jindo-flink-${version}-full.jar do diretório plugins/flink/ do JindoSDK para o diretório raiz lib do Apache Flink.
```
mv plugins/flink/jindo-flink-${version}-full.jar lib/
```

Importante

Se o Apache Flink tiver um conector OSS integrado, remova-o: exclua o arquivo flink-oss-fs-hadoop-${flink-version}.jar do subdiretório lib ou do caminho plugins/oss-fs-hadoop.
Depois de configure o JindoSDK, use o prefixo oss:// nos jobs do Flink para gravar dados em streaming no OSS-HDFS ou no OSS. O JindoSDK identifica automaticamente o armazenamento de destino.

Exemplos

configure as opções gerais.

Para gravar dados no OSS-HDFS com semântica exactly-once:
1. ative o checkpointing do Apache Flink.
  Código de exemplo:
  1. crie um StreamExecutionEnvironment:
```
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
```
  2. ative o checkpointing:
```
env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);
```
2. Use uma source de dados que ofereça suporte à retransmissão de dados, como o Kafka.
configure as opções rápidas.

especifique um caminho que comece com oss:// usando buckets e endpoints do OSS-HDFS. Nenhuma dependência adicional é necessária.
1. Adicione um sink.
  
  O código a seguir grava um DataStream<String> no 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
  O campo .{oss-hdfs-endpoint} é opcional. Caso seja omitido, especifique o endpoint correto do OSS-HDFS na configuração do Flink ou do Hadoop.
2. execute env.execute() para envie o job do Flink.

(Opcional) Defina as configurações personalizadas

Defina parâmetros personalizados ao envie jobs do Flink para ative recursos específicos.

Use -yD para configure 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ção no OSS-HDFS:

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

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