I/O gestiti da Dataflow per Apache Iceberg

Managed I/O supporta le seguenti funzionalità per Apache Iceberg:

Cataloghi
  • Hadoop
  • Hive
  • Cataloghi basati su REST
  • Metastore BigQuery (richiede Apache Beam SDK 2.62.0 o versioni successive se non utilizzi Portable Runner)
Leggi le funzionalità Lettura batch
Funzionalità di scrittura

Per le tabelle BigQuery per Apache Iceberg, utilizza il connettore BigQueryIO con l'API BigQuery Storage. La tabella deve già esistere; la creazione dinamica della tabella non è supportata.

Requisiti

I seguenti SDK supportano I/O gestito per Apache Iceberg:

  • SDK Apache Beam per Java versione 2.58.0 o successive
  • SDK Apache Beam per Python versione 2.61.0 o successive

Configurazione

Managed I/O per Apache Iceberg supporta i seguenti parametri di configurazione:

ICEBERG Leggi

Configurazione Tipo Descrizione
table str Identificatore della tabella Iceberg.
catalog_name str Nome del catalogo contenente la tabella.
catalog_properties map[str, str] Proprietà utilizzate per configurare il catalogo Iceberg.
config_properties map[str, str] Proprietà passate alla configurazione Hadoop.
drop list[str] Un sottoinsieme di nomi di colonne da escludere dalla lettura. Se è nullo o vuoto, verranno lette tutte le colonne.
filtro str Predicato simile a SQL per filtrare i dati al momento della scansione. Esempio: "id > 5 AND status = 'ACTIVE'". Utilizza la sintassi di Apache Calcite: https://calcite.apache.org/docs/reference.html
keep list[str] Un sottoinsieme di nomi di colonna da leggere esclusivamente. Se è nullo o vuoto, verranno lette tutte le colonne.

ICEBERG Scrittura

').

Configurazione Tipo Descrizione
table str Un identificatore di tabella completo. Puoi anche fornire un modello per scrivere in più destinazioni dinamiche, ad esempio: `dataset.my_{col1}_{col2.nested}_table`.
allowed_lateness_seconds int32 Per quanto tempo un record in ritardo può rimanere indietro rispetto alla filigrana prima di essere eliminato completamente, anziché essere indirizzato all'output dead_letter. Il valore predefinito è 21600 (6 ore). Al momento è supportato solo nella modalità "merge-on-read".
autosharding boolean Attiva lo sharding dinamico per regolare automaticamente il numero di writer paralleli in base al volume di dati. Gestisce la distorsione dei dati suddividendo ulteriormente le partizioni in più shard per evitare colli di bottiglia durante le scritture a velocità effettiva elevata. Disponibile solo con la modalità di distribuzione "hash".
catalog_name str Nome del catalogo contenente la tabella.
catalog_properties map[str, str] Proprietà utilizzate per configurare il catalogo Iceberg.
change_type_column str Unione in lettura. Il nome della colonna facoltativa che rappresenta il tipo di modifica della riga (INSERT, UPDATE_BEFORE, UPDATE_AFTER o DELETE). Questa colonna verrà rimossa dalla riga di dati prima della scrittura in Iceberg. Se non viene impostato, il sink utilizzerà ValueKind nativo dell'elemento
change_type_map map[str, str] Unione in lettura. (Facoltativo) Mappa da un valore change_type_column al nome canonico del tipo di modifica (vedi sopra).
config_properties map[str, str] Proprietà passate alla configurazione Hadoop.
direct_write_byte_limit int32 Per una pipeline di streaming, imposta il limite per il passaggio dei bundle al percorso di scrittura diretta.
distribution_mode str Definisce la distribuzione dei dati di scrittura. Distribuzioni supportate: - Nessuna: non mescolare le righe (impostazione predefinita) - Hash: mescola le righe in base alla chiave di partizione prima di scrivere i dati
drop list[str] Un elenco di nomi di campi da eliminare dal record di input prima della scrittura. Si esclude a vicenda con "keep" e "only". Nella modalità di unione in lettura, le colonne di controllo vengono sempre eliminate.
equality_columns list[str] Colonne che definiscono l'identità della riga (campi di eliminazione dell'uguaglianza). Il valore predefinito è costituito dai campi identificatori (chiave primaria) della tabella di destinazione. Obbligatorio se la tabella non esiste ancora. Al momento è supportato solo nella modalità "merge-on-read".
keep list[str] Un elenco di nomi di campi da conservare nel record di input. Tutti gli altri campi vengono eliminati prima della scrittura. Si esclude a vicenda con "drop" e "only". Nella modalità di unione in lettura, le colonne di controllo vengono eliminate, a meno che non siano elencate qui.
maximum_table_cache_size int32 Per una pipeline batch, imposta il numero massimo di specifiche dei metadati della tabella da memorizzare nella cache in memoria. Le tabelle che superano questo limite vengono caricate nel catalogo locale del worker.
modalità str Controlla la modalità di scrittura delle righe. "append" (aggiungi) aggiunge ogni riga come nuovi dati. "merge-on-read" considera ogni riga come una modifica (INSERT, UPDATE_BEFORE, UPDATE_AFTER o DELETE) applicata alla tabella in base alla chiave primaria.
num_shards int32 Il numero di shard hash della chiave primaria deterministici per destinazione, ovvero il parallelismo di scrittura massimo per destinazione. Un valore troppo basso può limitare le scritture, mentre un valore troppo alto può produrre più file. Il valore predefinito è 16. Al momento è supportato solo nella modalità "merge-on-read".
solo str Il nome di un singolo campo del record da scrivere. Si esclude a vicenda con "keep" e "drop".
partition_fields list[str] Campi utilizzati per creare una specifica di partizione che viene applicata quando vengono create le tabelle. Per un campo "foo", le trasformazioni di partizione disponibili sono:
  • foo
  • truncate(foo, N)
  • bucket(foo, N)
  • hour(foo)
  • day(foo)
  • month(foo)
  • year(foo)
  • void(foo)

Per saperne di più sulle trasformazioni delle partizioni, visita la pagina https://iceberg.apache.org/spec/#partition-transforms.

sequence_number_column str Unione in lettura. Il nome della colonna obbligatoria che rappresenta il numero di sequenza monotono utilizzato per ordinare le modifiche di una singola chiave. Il valore predefinito è "_commit_snapshot_sequence_number". Questa colonna verrà rimossa dalla riga di dati prima della scrittura in Iceberg.
shards_per_partition int32 Numero massimo di partizioni che possono occupare le righe di una singola partizione. Valori più bassi scrivono meno file per commit, ma riducono anche il parallelismo di scrittura per partizione. Un valore pari a 1 assegna ogni partizione a un writer. Ignorato per le tabelle non partizionate. Il valore deve essere compreso tra 1 e num_shards; il valore predefinito è num_shards. Al momento è supportato solo nella modalità "merge-on-read".
sink_id str Un identificatore stabile per questo sink, utilizzato per lo spazio dei nomi dei token di idempotenza scritti nel riepilogo dello snapshot Iceberg di ogni commit. Il valore predefinito è un UUID univoco per ogni scrittura. Impostalo in modo esplicito (e mantienilo stabile nei riavvii) per i commit esattamente una volta nei riavvii di una particolare scrittura di streaming. Un caricamento batch con un sink_id stabile viene eseguito il commit una sola volta (i caricamenti batch successivi con lo stesso sink_id vengono ignorati). Al momento è supportato solo nella modalità "merge-on-read".
snapshot_properties map[str, str] Proprietà chiave/valore aggiuntive da aggiungere al riepilogo dello snapshot Iceberg di ogni commit. Le chiavi con il prefisso "beam.cdc." sono riservate e rifiutate. Al momento è supportato solo nella modalità "merge-on-read".
sort_fields list[str] Campi utilizzati per impostare l'ordine di ordinamento della tabella, applicato quando viene creata la tabella. Ogni voce ha il formato <term> [asc|desc] [nulls first|nulls last], dove <term> è un nome di campo o una delle trasformazioni della partizione (ad es. bucket(col, 4), day(ts)). La direzione è predefinita su crescente; l'ordine dei valori null è predefinito su nulls-first per crescente e nulls-last per decrescente. Nota: questa operazione imposta l'ordine di ordinamento dichiarato della tabella come metadati; non fa sì che Beam ordini fisicamente i record prima della scrittura. Per ulteriori informazioni sugli ordini di ordinamento, visita la pagina https://iceberg.apache.org/spec/#sort-orders.
sorter_memory_mb int32 Le dimensioni del buffer in memoria (MB) per l'ordinamento pre-scrittura; i gruppi più grandi di questo vengono memorizzati sul disco. Deve essere >= 1. Il valore predefinito è 100. Al momento è supportato solo nella modalità "merge-on-read".
table_cache_polling_buckets int32 Imposta il numero di bucket/worker paralleli utilizzati per eseguire query sul catalogo Iceberg durante gli aggiornamenti. Il valore predefinito è 1.
table_cache_refresh_interval_seconds int32 Per una pipeline di streaming, imposta l'intervallo in secondi in cui i metadati della tabella vengono aggiornati dal catalogo.
table_properties map[str, str] Proprietà della tabella Iceberg da impostare sulla tabella al momento della creazione. Per ulteriori informazioni sulle proprietà delle tabelle, visita la pagina https://iceberg.apache.org/docs/latest/configuration/#table-properties.
token_heartbeat_seconds int32 Solo streaming. Se impostato, il sink emetterà un commit periodico di aggiornamento del token vuoto durante l'inattività, in modo che il thread dello snapshot con timestamp sink_id rimanga recente e sia meno probabile che venga perso a causa di expire_snapshots. Disattivato per impostazione predefinita. Al momento è supportato solo nella modalità "merge-on-read".
triggering_frequency_seconds int32 Per una pipeline di streaming, imposta la frequenza con cui vengono prodotti gli snapshot.
upsert boolean Unione in lettura. Se è vero, viene applicata solo l'immagine successiva di ogni modifica (INSERT/UPDATE_AFTER), come upsert; i record UPDATE_BEFORE vengono eliminati. Valore predefinito: false.
use_side_input_table_cache boolean Consente la memorizzazione nella cache degli input secondari con scadenza dei metadati della tabella Iceberg tra i worker per ridurre il carico del catalogo.
write_properties map[str, str] Proprietà applicate al writer di file sottostante (ad es. proprietà di scrittura Parquet come "write.parquet.bloom-filter-enabled.column.

Passaggi successivi

Per ulteriori informazioni ed esempi di codice, consulta i seguenti argomenti: