Realtime Compute for Apache Flink:Conector Apache Iceberg

O Apache Iceberg é um formato de tabela de data lake aberto. Use o Apache Iceberg para criar rapidamente seu próprio serviço de armazenamento de data lake em HDFS ou OSS na nuvem e utilize mecanismos de computação do ecossistema open source de big data, como Flink, Spark, Hive e Presto, para analisar dados no data lake.

O conector Apache Iceberg oferece os seguintes recursos:

• Cria um serviço de armazenamento de data lake leve e de baixo custo, baseado em HDFS ou armazenamento de objetos.

• Fornece semântica ACID completa.

• Suporta consultas time travel para acessar versões históricas dos dados.

• Filtra dados com eficiência.

• Suporta evolução de schema.

• Suporta evolução de partição.

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

Certifique-se de ter um bucket no OSS e um banco de dados no Data Lake Formation (DLF). Para mais informações, consulte Criar um bucket e Bancos de dados, tabelas e funções.

Este exemplo usa o conector Datagen para gerar dados streaming aleatórios e gravá-los em uma tabela Iceberg.

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;

Ingestão de dados

Use o conector Apache Iceberg como destino em um job YAML para ingestão de dados.

Sintaxe

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

Parâmetros

Reutilizar um catálogo existente

A partir do VVR 11.5, referencie diretamente um catálogo Iceberg integrado criado na página Data Management em um job de ingestão de dados Flink CDC. Isso simplifica a configuração ao reduzir o número de propriedades de conexão necessárias.

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

Jobs de ingestão de dados reutilizam automaticamente todos os parâmetros do catálogo Iceberg. Isso equivale a configurar manualmente os parâmetros com o prefixo catalog.properties. no job YAML.

Para substituir os parâmetros reutilizados, especifique explicitamente os parâmetros YAML correspondentes. Esses parâmetros têm precedência.

Exemplo

O exemplo a seguir mostra como usar um catálogo DLF como catálogo Iceberg e gravar dados no Data Lake Formation (DLF):

• Para informações sobre os parâmetros com prefixo catalog.properties, consulte Criar um catálogo Iceberg DLF.

  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

Alterações de schema

Quando usado como destino de ingestão de dados, o conector Apache Iceberg suporta as seguintes alterações de schema:

• CREATE TABLE

• ADD COLUMN

• ALTER COLUMN TYPE (não há suporte para modificação do tipo de coluna de chave primária)

• RENAME COLUMN

• DROP COLUMN

• TRUNCATE TABLE

• DROP TABLE