E/S gérées Dataflow pour Apache Iceberg

Les E/S gérées sont compatibles avec les fonctionnalités suivantes pour Apache Iceberg :

Catalogues
  • Hadoop
  • Hive
  • Catalogues basés sur REST
  • Métastore BigQuery (nécessite le SDK Apache Beam 2.62.0 ou version ultérieure si vous n'utilisez pas l'exécuteur v2)
Fonctionnalités de lecture Lecture par lot
Fonctionnalités d'écriture

Pour les tables BigQuery pour Apache Iceberg, utilisez le BigQueryIO connecteur avec l'API BigQuery Storage. La table doit déjà exister. La création de tables dynamiques n'est pas prise en charge.

Conditions requises

Les SDK suivants sont compatibles avec les E/S gérées pour Apache Iceberg :

  • SDK Apache Beam pour Java version 2.58.0 ou ultérieure
  • SDK Apache Beam pour Python version 2.61.0 ou ultérieure

Configuration

Les E/S gérées pour Apache Iceberg sont compatibles avec les paramètres de configuration suivants :

Lecture ICEBERG

Configuration Type Description
table str Identifiant de la table Iceberg.
catalog_name str Nom du catalogue contenant la table.
catalog_properties map[str, str] Propriétés utilisées pour configurer le catalogue Iceberg.
config_properties map[str, str] Propriétés transmises à la configuration Hadoop.
drop list[str] Sous-ensemble de noms de colonnes à exclure de la lecture. Si la valeur est nulle ou vide, toutes les colonnes seront lues.
filter str Prédicat de type SQL pour filtrer les données au moment de l'analyse. Exemple : "id > 5 AND status = 'ACTIVE'". Utilise la syntaxe Apache Calcite : https://calcite.apache.org/docs/reference.html
keep list[str] Sous-ensemble de noms de colonnes à lire exclusivement. Si la valeur est nulle ou vide, toutes les colonnes seront lues.

Écriture ICEBERG

Configuration Type Description
table str Identifiant de table complet. Vous pouvez également fournir un modèle pour écrire vers plusieurs destinations dynamiques, par exemple : `dataset.my_{col1}_{col2.nested}_table`.
autosharding boolean Active le sharding dynamique pour ajuster automatiquement le nombre d'écrivains parallèles en fonction du volume de données. Il gère l'asymétrie des données en subdivisant davantage les partitions en plusieurs fragments pour éviter les goulots d'étranglement lors des écritures à haut débit. Disponible uniquement avec le mode de distribution "hash".
catalog_name str Nom du catalogue contenant la table.
catalog_properties map[str, str] Propriétés utilisées pour configurer le catalogue Iceberg.
config_properties map[str, str] Propriétés transmises à la configuration Hadoop.
direct_write_byte_limit int32 Pour un pipeline de traitement en flux continu, définit la limite de transfert des lots vers le chemin d'écriture direct.
distribution_mode str Définit la distribution des données d'écriture. Distributions compatibles : - none : ne pas mélanger les lignes (par défaut) - hash : mélanger les lignes par clé de partition avant d’écrire les données
drop list[str] Liste des noms de champs à supprimer de l'enregistrement d'entrée avant l'écriture. Incompatible avec "keep" et "only".
keep list[str] Liste des noms de champs à conserver dans l'enregistrement d'entrée. Tous les autres champs sont supprimés avant l'écriture. Incompatible avec "drop" et "only".
only str Nom d'un champ d'enregistrement unique devant être écrit. Incompatible avec "keep" et "drop".
partition_fields list[str] Champs utilisés pour créer une spécification de partition qui est appliquée lors de la création des tables. Pour un champ "foo", les transformations de partition disponibles sont les suivantes :
  • foo
  • truncate(foo, N)
  • bucket(foo, N)
  • hour(foo)
  • day(foo)
  • month(foo)
  • year(foo)
  • void(foo)

Pour en savoir plus sur les transformations de partition, consultez https://iceberg.apache.org/spec/#partition-transforms.

sort_fields list[str] Champs utilisés pour définir l'ordre de tri de la table, appliqué lors de sa création. Chaque entrée se présente sous la forme <term> [asc|desc] [nulls first|nulls last], où <term> est un nom de champ ou l'une des transformations de partition (par exemple, bucket(col, 4), day(ts)). La direction est définie par défaut sur "ascendant", l'ordre des valeurs nulles est défini par défaut sur "nulls first" pour l'ordre ascendant et sur "nulls last" pour l'ordre descendant. Remarque : Cela définit l'ordre de tri déclaré de la table en tant que métadonnées. Cela n'entraîne pas le tri physique des enregistrements par Beam avant l'écriture. Pour en savoir plus sur les ordres de tri, consultez https://iceberg.apache.org/spec/#sort-orders.
table_properties map[str, str] Propriétés de la table Iceberg à définir sur la table lors de sa création. Pour en savoir plus sur les propriétés de table, consultez https://iceberg.apache.org/docs/latest/configuration/#table-properties.
triggering_frequency_seconds int32 Pour un pipeline de traitement en flux continu, définit la fréquence à laquelle les instantanés sont produits.

Étape suivante

Pour en savoir plus et obtenir des exemples de code, consultez les sujets suivants :