Realtime Compute for Apache Flink:Conector de Apache Iceberg

Apache Iceberg es un formato de tabla de data lake abierto. Puede usar Apache Iceberg para crear rápidamente su propio servicio de almacenamiento de data lake en HDFS o en OSS en la nube, y utilizar motores de cómputo del ecosistema de big data de código abierto, como Flink, Spark, Hive y Presto, para realizar análisis de data lake.

El conector de Apache Iceberg ofrece las siguientes funciones:

• Crea un servicio de almacenamiento de data lake ligero y de bajo costo basado en HDFS o object storage.

• Proporciona semántica ACID completa.

• Admite consultas de viaje en el tiempo para acceder a versiones históricas de los datos.

• Permite un filtrado eficiente de datos.

• Admite la evolución de esquemas.

• Admite la evolución de particiones.

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

Asegúrese de contar con un bucket de OSS y una base de datos de Data Lake Formation (DLF). Para obtener más información, consulte Create a bucket y Databases, tables, and functions.

Este ejemplo utiliza el conector Datagen para generar datos streaming aleatorios y escribirlos en una tabla de 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;

Ingesta de datos

Puede utilizar el conector de Apache Iceberg como destino en un trabajo YAML para la ingesta de datos.

Sintaxis

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 un catálogo existente

A partir de VVR 11.5, puede hacer referencia directa a un catálogo integrado de Iceberg creado en la página Data Management dentro de un trabajo de ingesta de datos de Flink CDC. Esto simplifica su configuración al reducir la cantidad de propiedades de conexión obligatorias.

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

Los trabajos de ingesta de datos pueden reutilizar automáticamente todos los parámetros del catálogo de Iceberg. Esto equivale a configurar manualmente los parámetros con el prefijo catalog.properties. en su trabajo YAML.

Si desea anular los parámetros reutilizados, puede especificar explícitamente los parámetros YAML correspondientes. Estos parámetros tendrán prioridad.

Ejemplo

A continuación se muestra un ejemplo de cómo usar un catálogo DLF como catálogo de Iceberg y escribir datos en Data Lake Formation (DLF):

• Para obtener información sobre los parámetros con el prefijo catalog.properties, consulte Create an Iceberg DLF catalog.

  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

Cambios de esquema

Cuando se utiliza como destino de ingesta de datos, el conector de Apache Iceberg admite los siguientes cambios de esquema:

• CREATE TABLE

• ADD COLUMN

• ALTER COLUMN TYPE (No se admite la modificación del tipo de una columna de clave principal).

• RENAME COLUMN

• DROP COLUMN

• TRUNCATE TABLE

• DROP TABLE