Realtime Compute for Apache Flink:Apache Iceberg bağlayıcısı

Apache Iceberg açık kaynaklı bir veri gölü tablo biçimidir. Apache Iceberg kullanarak HDFS veya cloud OSS üzerinde kendi veri gölü depolama hizmetinizi hızla oluşturabilir; Flink, Spark, Hive ve Presto gibi açık kaynaklı büyük veri ekosisteminden gelen işlem motorlarını kullanarak veri gölü analizi yapabilirsiniz.

Apache Iceberg bağlayıcısı aşağıdaki özellikleri sunar:

• HDFS veya nesne depolama alanı üzerine düşük maliyetli, hafif bir veri gölü depolama hizmeti oluşturur.

• Tam ACID semantiği sunar.

• Geçmiş veri sürümlerine erişmek için zaman yolculuğu sorgularını destekler.

• Verimli veri filtrelemeyi destekler.

• Şema evrimini destekler.

• Bölümleme evrimini destekler.

CREATE TABLE iceberg_table (
  id    BIGINT,
  data  STRING
  PRIMARY KEY(`id`) NOT ENFORCED
)
 PARTITIONED BY (data)
 WITH (
 'connector' = 'iceberg',
  ...
);

Bir OSS paketiniz ve bir Data Lake Formation (DLF) veritabanınız olduğundan emin olun. Daha fazla bilgi için bkz. Paket oluşturma ve Veritabanları, tablolar ve işlevler.

Bu örnekte, Datagen bağlayıcısı kullanılarak rastgele akış verileri oluşturulur ve veriler bir Iceberg tablosuna yazılır.

CREATE TEMPORARY TABLE datagen(
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE dlf_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

INSERT INTO dlf_iceberg SELECT * FROM datagen;

Veri alımı

Apache Iceberg bağlayıcısını, veri alımı için bir YAML işinde hedef olarak kullanabilirsiniz.

Söz dizimi

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.rest.signing-region: cn-beijing
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
  catalog.properties.warehouse: flink_iceberg
  catalog.properties.type: rest
  catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

Parametreler

Mevcut bir kataloğu yeniden kullanma

VVR 11.5'ten başlayarak, bir Flink CDC veri alımı işinde Veri Yönetimi sayfasında oluşturulan yerleşik bir Iceberg kataloğuna doğrudan başvurabilirsiniz. Bu yaklaşım, gerekli bağlantı özelliği sayısını azaltarak yapılandırmanızı basitleştirir.

sink:
  type: iceberg
  using.built-in-catalog: iceberg_catalog

Veri alımı işleri, tüm Iceberg katalog parametrelerini otomatik olarak yeniden kullanabilir. Bu durum, YAML işinizde catalog.properties. önekiyle parametreleri manuel olarak yapılandırmaya eşdeğerdir.

Yeniden kullanılan parametreleri geçersiz kılmak istiyorsanız ilgili YAML parametrelerini açıkça belirtebilirsiniz. Bu parametreler önceliklidir.

Örnek

Aşağıdaki örnekte, bir DLF kataloğunun Iceberg kataloğu olarak nasıl kullanılacağı ve verilerin Data Lake Formation (DLF) adresine nasıl yazılacağı gösterilmektedir:

• catalog.properties önekiyle başlayan parametreler hakkında bilgi için bkz. Iceberg DLF kataloğu oluşturma.

  source:
    type: mysql
    name: MySQL Source
    hostname: ${secret_values.mysql.hostname}
    port: ${mysql.port}
    username: ${secret_values.mysql.username}
    password: ${secret_values.mysql.password}
    tables: ${mysql.source.table}
    server-id: 8601-8604
  
  sink:
    type: iceberg
    name: Iceberg Sink
    catalog.properties.rest.signing-region: cn-beijing
    catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
    catalog.properties.warehouse: flink_iceberg
    catalog.properties.type: rest
    catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

Şema değişiklikleri

Veri alımı hedefi olarak kullanıldığında Apache Iceberg bağlayıcısı aşağıdaki şema değişikliklerini destekler:

• CREATE TABLE

• ADD COLUMN

• ALTER COLUMN TYPE (Birincil anahtar sütununun türünün değiştirilmesi desteklenmez.)

• RENAME COLUMN

• DROP COLUMN

• TRUNCATE TABLE

• DROP TABLE