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

Açık kaynaklı Apache Flink, OSS-HDFS'e akış yazmayı veya depolama ortamları için tam olarak bir kez (exactly-once) semantiğini desteklemez. Bu yetenekleri etkinleştirmek için Apache Flink ile birlikte JindoSDK kullanın.

Not

JindoSDK'yi dağıtmadan akış verilerini OSS-HDFS'e yazmak için Realtime Compute for Apache Flink kullanın. Daha fazla bilgi için bkz. Realtime Compute for Apache Flink kullanarak OSS veya OSS-HDFS'ten okuma veya bu hizmetlere yazma.

Ön koşullar

Bir ECS örneği oluşturun. Örnek oluşturma.
Uygun erişim izinlerinin verildiği bir paket için OSS-HDFS'i etkinleştirin. Daha fazla bilgi için bkz. OSS-HDFS hizmetini etkinleştirme.
Apache Flink 1.10.1 veya sonraki bir sürümü yükleyin. Flink 1.16.0 ve sonraki sürümler doğrulanmamıştır. Flink'i Apache Flink adresinden indirin.

JindoSDK'yi yapılandırma

ECS örneğinde oturum açın. Örneğe bağlanma.
JindoSDK JAR paketinin en son sürümünü indirin ve arşivden çıkarın. JindoSDK'yı indirmek için GitHub adresini ziyaret edin.
JindoSDK'deki plugins/flink/ dizininden jindo-flink-${version}-full.jar dosyasını Apache Flink kök lib dizinine taşıyın.
```
mv plugins/flink/jindo-flink-${version}-full.jar lib/
```

Önemli

Apache Flink yerleşik bir OSS bağlayıcısına sahipse, lib alt dizininden veya plugins/oss-fs-hadoop yolundan flink-oss-fs-hadoop-${flink-version}.jar dosyasını silerek bu bağlayıcıyı kaldırın.
JindoSDK'yi yapılandırdıktan sonra, Flink işlerinde akış verilerini OSS-HDFS'e veya OSS'e yazmak için oss:// ön ekini kullanın. JindoSDK hedef depolamayı otomatik olarak tanımlar.

Örnekler

Genel ayarları yapılandırın.

Verileri tam olarak bir kez (exactly-once) semantiği ile OSS-HDFS'e yazmak için:
1. Apache Flink kontrol noktasını (checkpointing) etkinleştirin.
  Örnek kod:
  1. Bir StreamExecutionEnvironment oluşturun:
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. Kontrol noktasını etkinleştirin:
    env.enableCheckpointing(<userDefinedCheckpointInterval>, CheckpointingMode.EXACTLY_ONCE);
2. Kafka gibi veri yeniden iletimini destekleyen bir veri kaynağı kullanın.
Hızlı ayarları yapılandırın.

OSS-HDFS paketleri ve uç noktaları ile oss:// ile başlayan bir yol belirtin. Ek bağımlılıklar gerekmez.
1. Bir havuz (sink) ekleyin.
  
  Aşağıdaki kod, bir DataStream<String> öğesini OSS-HDFS'e yazar:
```
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);
```
  Önemli
  .{oss-hdfs-endpoint} alanı isteğe bağlıdır. Atlanırsa, Flink veya Hadoop yapılandırmasında doğru OSS-HDFS uç noktasını belirtin.
2. Flink işini göndermek için env.execute() komutunu çalıştırın.

(İsteğe bağlı) Özel ayarları yapılandırma

Belirli özellikleri etkinleştirmek için Flink işlerini gönderirken özel parametreler yapılandırabilirsiniz.

YARN tabanlı Flink iş gönderimlerini yapılandırmak için -yD seçeneğini kullanın:

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

Entropi enjeksiyonu, hedef yoldaki belirtilen bir dizeyi rastgele bir dizeyle değiştirerek yazma işlemlerini bölümlere dağıtır ve performansı artırır.

OSS-HDFS yazma işlemleri için aşağıdaki parametreleri yapılandırın:

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

Yazma işlemleri sırasında, yoldaki <user-defined-key> dizesi, uzunluğu <user-defined-length> değerine eşit olan rastgele bir dizeyle değiştirilir. <user-defined-length> değeri 0'dan büyük olmalıdır.