E/S gerenciada pelo Dataflow para o Apache Iceberg

A E/S gerenciada oferece suporte aos seguintes recursos do Apache Iceberg:

Catálogos
  • Hadoop
  • Hive
  • Catálogos baseados em REST
  • Metastore do BigQuery (requer o SDK do Apache Beam 2.62.0 ou mais recente se não estiver usando o Portable Runner)
Ler recursos Leitura em lote
Capacidades de gravação

Para tabelas do BigQuery para Apache Iceberg, use o conector BigQueryIO com a API BigQuery Storage. A tabela já precisa existir. Não é possível criar tabelas dinâmicas.

Requisitos

Os seguintes SDKs são compatíveis com a E/S gerenciada para o Apache Iceberg:

  • SDK do Apache Beam para Java versão 2.58.0 ou mais recente
  • SDK do Apache Beam para Python versão 2.61.0 ou mais recente

Configuração

A E/S gerenciada para Apache Iceberg é compatível com os seguintes parâmetros de configuração:

ICEBERG Lidos

Configuração Tipo Descrição
table str Identificador da tabela do Iceberg.
catalog_name str O nome do catálogo que contém a tabela.
catalog_properties map[str, str] Propriedades usadas para configurar o catálogo do Iceberg.
config_properties map[str, str] Propriedades transmitidas para a configuração do Hadoop.
drop list[str] Um subconjunto de nomes de colunas a serem excluídos da leitura. Se for nulo ou vazio, todas as colunas serão lidas.
filtro str Predicado semelhante a SQL para filtrar dados no momento da verificação. Exemplo: "id > 5 AND status = 'ACTIVE'". Usa a sintaxe do Apache Calcite: https://calcite.apache.org/docs/reference.html
keep list[str] Um subconjunto de nomes de colunas a serem lidos exclusivamente. Se for nulo ou vazio, todas as colunas serão lidas.

ICEBERG Gravar

').

Configuração Tipo Descrição
table str Um identificador de tabela totalmente qualificado. Você também pode fornecer um modelo para gravar em vários destinos dinâmicos, por exemplo: `dataset.my_{col1}_{col2.nested}_table`.
allowed_lateness_seconds int32 Quanto tempo um registro atrasado pode ficar atrás da marca d'água antes de ser descartado completamente, em vez de ser encaminhado para a saída dead_letter. O padrão é 21600 (6 horas). No momento, só há suporte no modo "merge-on-read".
fragmentação automática boolean Ativa o fragmentação dinâmica para ajustar automaticamente o número de gravadores paralelos com base no volume de dados. Ele processa a distorção de dados subdividindo ainda mais as partições em vários fragmentos para evitar gargalos durante gravações de alta capacidade de processamento. Disponível apenas com o modo de distribuição "hash".
catalog_name str O nome do catálogo que contém a tabela.
catalog_properties map[str, str] Propriedades usadas para configurar o catálogo do Iceberg.
change_type_column str Somente mesclagem na leitura. O nome da coluna opcional que representa o tipo de mudança da linha (INSERT, UPDATE_BEFORE, UPDATE_AFTER ou DELETE). Essa coluna será removida da linha de dados antes da gravação no Iceberg. Se não for definido, o gravador vai usar o ValueKind nativo do elemento.
change_type_map map[str, str] Somente mesclagem na leitura. Mapa opcional de um valor change_type_column para o nome canônico do tipo de mudança (consulte acima).
config_properties map[str, str] Propriedades transmitidas para a configuração do Hadoop.
direct_write_byte_limit int32 Em um pipeline de streaming, define o limite para elevar pacotes para o caminho de gravação direta.
distribution_mode str Define a distribuição dos dados de gravação. Distribuições compatíveis: - nenhuma: não embaralha as linhas (padrão) - hash: embaralha as linhas por chave de partição antes de gravar os dados
drop list[str] Uma lista de nomes de campos a serem descartados do registro de entrada antes da gravação. É mutuamente exclusivo com "keep" e "only". No modo de mesclagem na leitura, as colunas de controle são sempre descartadas.
equality_columns list[str] Colunas que definem a identidade da linha (campos de exclusão por igualdade). O padrão são os campos de identificador (chave primária) da tabela de destino. Obrigatório se a tabela ainda não existir. No momento, só há suporte no modo "merge-on-read".
keep list[str] Uma lista de nomes de campos a serem mantidos no registro de entrada. Todos os outros campos são descartados antes da gravação. É mutuamente exclusivo com "drop" e "only". No modo de mesclagem na leitura, as colunas de controle são descartadas, a menos que estejam listadas aqui.
maximum_table_cache_size int32 Para um pipeline em lote, define o número máximo de especificações de metadados da tabela a serem armazenadas em cache na memória. As tabelas que excedem esse limite voltam ao carregamento do catálogo local do worker.
modo str Controla como as linhas são gravadas. "append" (padrão) anexa todas as linhas como novos dados. "merge-on-read" trata cada linha como uma mudança (INSERT, UPDATE_BEFORE, UPDATE_AFTER ou DELETE) aplicada à tabela por chave primária.
num_shards int32 O número de fragmentos determinísticos de hash de chave primária por destino, ou seja, o paralelismo máximo de gravação por destino. Se for muito baixo, pode causar um gargalo nas gravações. Se for muito alto, pode gerar mais arquivos. O padrão é 16. No momento, só há suporte no modo "merge-on-read".
somente str O nome de um único campo de registro que deve ser gravado. É mutuamente exclusivo com "keep" e "drop".
partition_fields list[str] Campos usados para criar uma especificação de partição que é aplicada quando as tabelas são criadas. Para um campo "foo", as transformações de partição disponíveis são:
  • foo
  • truncate(foo, N)
  • bucket(foo, N)
  • hour(foo)
  • day(foo)
  • month(foo)
  • year(foo)
  • void(foo)

Para mais informações sobre transformações de partição, acesse https://iceberg.apache.org/spec/#partition-transforms.

sequence_number_column str Somente mesclagem na leitura. O nome da coluna obrigatória que representa o número de sequência monotônica usado para ordenar as mudanças de uma única chave. O padrão é "_commit_snapshot_sequence_number". Essa coluna será removida da linha de dados antes da gravação no Iceberg.
shards_per_partition int32 Número máximo de fragmentos que as linhas de uma única partição podem ocupar. Valores mais baixos gravam menos arquivos por commit, mas também reduzem o paralelismo de gravação por partição. Um valor de 1 fixa cada partição a um gravador. Ignorado para tabelas não particionadas. Precisa estar entre 1 e num_shards. O padrão é num_shards. No momento, só há suporte no modo "merge-on-read".
sink_id str Um identificador estável para esse coletor, usado para criar namespaces dos tokens de idempotência gravados no resumo do snapshot do Iceberg de cada commit. O padrão é um UUID exclusivo por gravação. Defina explicitamente (e mantenha estável em todas as reinicializações) para commits exatamente uma vez em todas as reinicializações de uma gravação de streaming específica. Um carregamento em lote com um sink_id estável é confirmado apenas uma vez. Os carregamentos em lote posteriores com o mesmo sink_id são ignorados. No momento, só há suporte no modo "merge-on-read".
snapshot_properties map[str, str] Propriedades extras de chave/valor a serem adicionadas ao resumo do snapshot do Iceberg de cada commit. As chaves com o prefixo "beam.cdc." são reservadas e rejeitadas. No momento, só há suporte no modo "merge-on-read".
sort_fields list[str] Campos usados para definir a ordem de classificação da tabela, aplicada quando ela é criada. Cada entrada tem a forma <term> [asc|desc] [nulls first|nulls last], em que <term> é um nome de campo ou uma das transformações de partição (por exemplo, bucket(col, 4), day(ts)). A direção padrão é crescente, e a ordem nula é "nulos primeiro" para crescente e "nulos por último" para decrescente. Observação: isso define a ordem de classificação declarada da tabela como metadados, mas não faz com que o Beam classifique fisicamente os registros antes de gravar. Para mais informações sobre ordens de classificação, acesse https://iceberg.apache.org/spec/#sort-orders (em inglês).
sorter_memory_mb int32 O tamanho do buffer na memória (MB) para a classificação pré-gravação. Grupos maiores que esse valor são gravados no disco. Precisa ser >= 1. O padrão é 100. No momento, só há suporte no modo "merge-on-read".
table_cache_polling_buckets int32 Define o número de intervalos/workers paralelos usados para consultar o catálogo do Iceberg durante as atualizações. O padrão é 1.
table_cache_refresh_interval_seconds int32 Para um pipeline de streaming, define o intervalo em segundos em que os metadados da tabela são atualizados do catálogo.
table_properties map[str, str] Propriedades da tabela do Iceberg que serão definidas nela quando ela for criada. Para mais informações sobre propriedades de tabela, acesse https://iceberg.apache.org/docs/latest/configuration/#table-properties.
token_heartbeat_seconds int32 Somente streaming. Se definido, o coletor vai emitir um commit periódico de atualização de token vazio enquanto estiver inativo. Assim, a linha de execução do snapshot carimbado com sink_id permanece recente e tem menos chances de ser perdida para expire_snapshots. Essa configuração está desativada por padrão. No momento, só há suporte no modo "merge-on-read".
triggering_frequency_seconds int32 Para um pipeline de streaming, define a frequência com que os snapshots são produzidos.
upsert boolean Somente mesclagem na leitura. Se for "true", somente a imagem após cada mudança (INSERT/UPDATE_AFTER) será aplicada como um upsert. Os registros UPDATE_BEFORE serão descartados. Padrão: false.
use_side_input_table_cache boolean Ativa o armazenamento em cache de entrada secundária expirável dos metadados da tabela do Iceberg em todos os workers para reduzir a carga do catálogo.
write_properties map[str, str] Propriedades aplicadas ao gravador de arquivos subjacente (por exemplo, propriedades de gravação do Parquet, como "write.parquet.bloom-filter-enabled.column.

A seguir

Para mais informações e exemplos de código, consulte os seguintes tópicos: