E/S administrada de Dataflow para Apache Iceberg

La E/S administrada admite las siguientes capacidades para Apache Iceberg:

Catálogos
  • Hadoop
  • Hive
  • Catálogos basados en REST
  • Metastore de BigQuery (requiere el SDK de Apache Beam 2.62.0 o una versión posterior si no se usa el Runner portátil)
Capacidades de lectura Lectura por lotes
Capacidades de escritura

Para las tablas de BigQuery para Apache Iceberg, usa el conector BigQueryIO con la API de BigQuery Storage. La tabla ya debe existir, ya que no se admite la creación de tablas dinámicas.

Requisitos

Los siguientes SDKs admiten la E/S administrada para Apache Iceberg:

  • SDK de Apache Beam para Java, versión 2.58.0 o posterior
  • SDK de Apache Beam para Python, versión 2.61.0 o posterior

Configuración

La E/S administrada para Apache Iceberg admite los siguientes parámetros de configuración:

ICEBERG Leer

Configuración Tipo Descripción
table str Es el identificador de la tabla de Iceberg.
catalog_name str Nombre del catálogo que contiene la tabla.
catalog_properties map[str, str] Propiedades que se usan para configurar el catálogo de Iceberg.
config_properties map[str, str] Propiedades que se pasan a la configuración de Hadoop.
drop list[str] Es un subconjunto de nombres de columnas que se excluirán de la lectura. Si es nulo o está vacío, se leerán todas las columnas.
filtrar str Es un predicado similar a SQL para filtrar datos durante el análisis. Ejemplo: "id > 5 AND status = 'ACTIVE'". Usa la sintaxis de Apache Calcite: https://calcite.apache.org/docs/reference.html
keep list[str] Es un subconjunto de nombres de columnas para leer exclusivamente. Si es nulo o está vacío, se leerán todas las columnas.

ICEBERG Escritura

').

Configuración Tipo Descripción
table str Un identificador de tabla completamente calificado. También puedes proporcionar una plantilla para escribir en varios destinos dinámicos, por ejemplo: `dataset.my_{col1}_{col2.nested}_table`.
allowed_lateness_seconds int32 Cuánto tiempo puede retrasarse un registro tardío con respecto a la marca de agua antes de que se descarte por completo, en lugar de enrutarse a la salida dead_letter. El valor predeterminado es 21600 (6 horas). Actualmente, solo se admite en el modo "merge-on-read".
Autosharding boolean Habilita el fragmentado dinámico para ajustar automáticamente la cantidad de escritores paralelos según el volumen de datos. Maneja la asimetría de los datos subdividiendo aún más las particiones en varios fragmentos para evitar cuellos de botella durante las escrituras de alta capacidad de procesamiento. Solo está disponible con el modo de distribución "hash".
catalog_name str Nombre del catálogo que contiene la tabla.
catalog_properties map[str, str] Propiedades que se usan para configurar el catálogo de Iceberg.
change_type_column str Solo se puede combinar en la lectura. Nombre de la columna opcional que representa el tipo de cambio de la fila (INSERT, UPDATE_BEFORE, UPDATE_AFTER o DELETE). Esta columna se quitará de la fila de datos antes de escribir en Iceberg. Si no se establece, el receptor usará el ValueKind nativo del elemento.
change_type_map map[str, str] Solo se puede combinar en la lectura. Mapa opcional de un valor de change_type_column al nombre canónico del tipo de cambio (consulta más arriba).
config_properties map[str, str] Propiedades que se pasan a la configuración de Hadoop.
direct_write_byte_limit int32 Para una canalización de transmisión, establece el límite para pasar los paquetes a la ruta de escritura directa.
distribution_mode str Define la distribución de los datos de escritura. Distribuciones admitidas: - none: No se mezclan las filas (opción predeterminada) - hash: Se mezclan las filas por clave de partición antes de escribir los datos
drop list[str] Es una lista de nombres de campos que se deben descartar del registro de entrada antes de la escritura. Es mutuamente excluyente con "keep" y "only". En el modo de combinación en lectura, las columnas de control siempre se descartan.
equality_columns list[str] Columnas que definen la identidad de la fila (campos de igualdad-borrado). El valor predeterminado son los campos de identificador (clave primaria) de la tabla de destino. Es obligatorio si la tabla aún no existe. Actualmente, solo se admite en el modo "merge-on-read".
keep list[str] Es una lista de nombres de campos que se conservarán en el registro de entrada. Se descartan todos los demás campos antes de la escritura. Es mutuamente excluyente con "drop" y "only". En el modo de combinación en lectura, las columnas de control se descartan, a menos que se indiquen aquí.
maximum_table_cache_size int32 Para una canalización por lotes, establece la cantidad máxima de especificaciones de metadatos de la tabla que se almacenarán en caché en la memoria. Las tablas que superen este límite volverán a la carga del catálogo local del trabajador.
Standard str Controla cómo se escriben las filas. "append" (predeterminado) agrega cada fila como datos nuevos. "merge-on-read" trata cada fila como un cambio (INSERT, UPDATE_BEFORE, UPDATE_AFTER o DELETE) aplicado a la tabla por clave primaria.
num_shards int32 Es la cantidad de fragmentos determinísticos de hash de clave principal por destino, es decir, el paralelismo máximo de escritura por destino. Si es demasiado bajo, puede generar un cuello de botella en las escrituras, y si es demasiado alto, puede producir más archivos. La configuración predeterminada es 16. Actualmente, solo se admite en el modo "merge-on-read".
solo str Es el nombre de un solo campo de registro que se debe escribir. Es mutuamente excluyente con "keep" y "drop".
partition_fields list[str] Son los campos que se usan para crear una especificación de partición que se aplica cuando se crean las tablas. Para un campo "foo", las transformaciones de partición disponibles son las siguientes:
  • foo
  • truncate(foo, N)
  • bucket(foo, N)
  • hour(foo)
  • day(foo)
  • month(foo)
  • year(foo)
  • void(foo)

Para obtener más información sobre las transformaciones de partición, visita https://iceberg.apache.org/spec/#partition-transforms.

sequence_number_column str Solo se puede combinar en la lectura. Es el nombre de la columna obligatoria que representa el número de secuencia monotónico que se usa para ordenar los cambios de una sola clave. El valor predeterminado es “_commit_snapshot_sequence_number”. Esta columna se quitará de la fila de datos antes de escribir en Iceberg.
shards_per_partition int32 Es la cantidad máxima de fragmentos que pueden ocupar las filas de una sola partición. Los valores más bajos escriben menos archivos por confirmación, pero también reducen el paralelismo de escritura por partición. Un valor de 1 fija cada partición a un escritor. Se ignora para las tablas no particionadas. Debe estar entre 1 y num_shards. El valor predeterminado es num_shards. Actualmente, solo se admite en el modo "merge-on-read".
sink_id str Es un identificador estable para este receptor, que se usa como espacio de nombres para los tokens de idempotencia escritos en el resumen de la instantánea de Iceberg de cada confirmación. El valor predeterminado es un UUID único por escritura. Establécelo de forma explícita (y mantenlo estable en los reinicios) para las confirmaciones de exactamente una vez en los reinicios de una escritura de transmisión en particular. Una carga por lotes con un sink_id estable se confirma solo una vez (se omiten las cargas por lotes posteriores con el mismo sink_id). Actualmente, solo se admite en el modo "merge-on-read".
snapshot_properties map[str, str] Son propiedades adicionales de clave/valor que se agregan al resumen de la instantánea de Iceberg de cada confirmación. Las claves con el prefijo "beam.cdc." están reservadas y se rechazan. Actualmente, solo se admite en el modo "merge-on-read".
sort_fields list[str] Son los campos que se usan para establecer el orden de clasificación de la tabla, que se aplica cuando se crea la tabla. Cada entrada tiene la forma <term> [asc|desc] [nulls first|nulls last], en la que <term> es un nombre de campo o una de las transformaciones de partición (p.ej., bucket(col, 4), day(ts)). La dirección predeterminada es ascendente; el orden de nulos predeterminado es nulos primero para el orden ascendente y nulos al final para el orden descendente. Nota: Esto establece el orden de clasificación declarado de la tabla como metadatos; no hace que Beam ordene físicamente los registros antes de escribir. Para obtener más información sobre los órdenes de clasificación, visita https://iceberg.apache.org/spec/#sort-orders.
sorter_memory_mb int32 Tamaño del búfer en memoria (MB) para la ordenación previa a la escritura; los grupos más grandes que este se vuelcan en el disco. Debe ser >= 1. La configuración predeterminada es 100. Actualmente, solo se admite en el modo "merge-on-read".
table_cache_polling_buckets int32 Establece la cantidad de buckets o trabajadores paralelos que se usan para consultar el catálogo de Iceberg durante las actualizaciones. El valor predeterminado es 1.
table_cache_refresh_interval_seconds int32 Para una canalización de transmisión, establece el intervalo en segundos en el que se actualizan los metadatos de la tabla desde el catálogo.
table_properties map[str, str] Propiedades de la tabla de Iceberg que se establecerán en la tabla cuando se cree. Para obtener más información sobre las propiedades de la tabla, visita https://iceberg.apache.org/docs/latest/configuration/#table-properties.
token_heartbeat_seconds int32 Solo transmisión. Si se configura, el receptor emitirá una confirmación periódica de actualización de token vacía mientras esté inactivo, de modo que su instantánea sellada con sink_id se mantenga reciente y sea menos probable que se pierda por expire_snapshots. Está inhabilitado de forma predeterminada. Actualmente, solo se admite en el modo "merge-on-read".
triggering_frequency_seconds int32 Para una canalización de transmisión, establece la frecuencia con la que se producen las instantáneas.
upsert boolean Solo se puede combinar en la lectura. Si es verdadero, solo se aplica la imagen posterior de cada cambio (INSERT/UPDATE_AFTER) como una operación de upsert; se descartan los registros de UPDATE_BEFORE. Valor predeterminado: false.
use_side_input_table_cache boolean Habilita el almacenamiento en caché de entradas complementarias con vencimiento de los metadatos de la tabla de Iceberg en los diferentes trabajadores para reducir la carga del catálogo.
write_properties map[str, str] Son las propiedades que se aplican al escritor de archivos subyacente (p.ej., propiedades de escritura de Parquet como "write.parquet.bloom-filter-enabled.column").

¿Qué sigue?

Para obtener más información y ejemplos de código, consulta los siguientes temas: